metube/app/subscriptions.py

"""Channel/playlist subscriptions: periodic yt-dlp flat extract + queue new videos."""

from __future__ import annotations

import asyncio
import copy
import logging
import os
import re
import time
import types
import uuid
from dataclasses import dataclass, field, fields
from functools import partial
from typing import Any, Optional

import yt_dlp
import yt_dlp.networking.impersonate
import bg_tasks
from dl_formats import merge_ytdl_option_layers
from state_store import AtomicJsonStore, read_legacy_shelf
from url_guard import validate_url, download_proxies

log = logging.getLogger("subscriptions")

# How many subscription feeds to scan at once. Bounded so one slow/hung feed
# doesn't serialize the rest, without bursting a large subscription list at the
# extractor (which risks rate-limiting / bot detection).
_MAX_CONCURRENT_CHECKS = 4

VIDEO_ONLY_MSG = (
    "This URL points to a single video, not a channel or playlist. Use Download instead."
)
_MEDIA_HINT_FIELDS = (
    "duration",
    "timestamp",
    "release_timestamp",
    "upload_date",
    "view_count",
    "live_status",
    "availability",
)


def _impersonate_opt(ytdl_options: dict) -> dict:
    opts = dict(ytdl_options)
    if "impersonate" in opts:
        opts["impersonate"] = yt_dlp.networking.impersonate.ImpersonateTarget.from_str(
            opts["impersonate"]
        )
    return opts


def _build_ydl_params(
    config, *, playlistend: Optional[int] = None, extra_opts: Optional[dict[str, Any]] = None
) -> dict:
    params: dict[str, Any] = {
        "quiet": not logging.getLogger().isEnabledFor(logging.DEBUG),
        "verbose": logging.getLogger().isEnabledFor(logging.DEBUG),
        "no_color": True,
        "extract_flat": True,
        "ignore_no_formats_error": True,
        "lazy_playlist": True,
        "paths": {"home": config.DOWNLOAD_DIR, "temp": config.TEMP_DIR},
        **config.YTDL_OPTIONS,
        **(extra_opts or {}),
        # A scan is a poll, not an add: it runs on a timer and queues items
        # through the download queue, which writes the feed metadata itself.
        # yt-dlp emits the playlist-level infojson/description/thumbnail
        # regardless of `download`, so without this a writeinfojson user would
        # get those files rewritten on every check interval. See issue #1040.
        "allow_playlist_files": False,
    }
    params = _impersonate_opt(params)
    if playlistend is not None and playlistend > 0:
        params["playlistend"] = playlistend
    return params


def _is_media_entry(entry: Any) -> bool:
    if not isinstance(entry, dict):
        return False
    etype = str(entry.get("_type") or "")
    if etype in ("playlist", "multi_video", "channel"):
        return False
    if entry.get("entries"):
        return False
    url = _entry_video_url(entry)
    if not url:
        return False
    ie_key = str(entry.get("ie_key") or entry.get("extractor_key") or "").lower()
    if any(token in ie_key for token in ("playlist", "channel", "tab")):
        return any(entry.get(field) is not None for field in _MEDIA_HINT_FIELDS)
    return True


def extract_flat_playlist(
    config, url: str, playlistend: int, *, extra_opts: Optional[dict[str, Any]] = None, _depth: int = 0
):
    """Return (info_dict, entries_list) for playlist/channel URLs."""
    params = _build_ydl_params(config, playlistend=playlistend, extra_opts=extra_opts)
    with yt_dlp.YoutubeDL(params=params) as ydl:
        info = ydl.extract_info(url, download=False)
    if not info:
        return None, []
    etype = info.get("_type") or "video"
    if etype == "video":
        return info, []
    if etype in ("playlist", "channel"):
        entries = info.get("entries") or []
        if isinstance(entries, types.GeneratorType):
            entries = list(entries)
        # Drop None placeholders from incomplete flat playlists
        entries = [e for e in entries if e]
        media_entries = [e for e in entries if _is_media_entry(e)]
        if media_entries:
            return info, media_entries
        if _depth < 1:
            proxies = download_proxies({**config.YTDL_OPTIONS, **(extra_opts or {})})
            for ent in entries[:5]:
                nested_url = _entry_video_url(ent)
                if not nested_url:
                    continue
                # nested_url comes from remote playlist content; guard it too,
                # against the same proxy map this scan is using.
                if validate_url(
                    nested_url,
                    allow_private=getattr(config, "ALLOW_PRIVATE_ADDRESSES", False),
                    proxies=proxies,
                ) is not None:
                    continue
                nested_info, nested_entries = extract_flat_playlist(
                    config,
                    nested_url,
                    playlistend,
                    extra_opts=extra_opts,
                    _depth=_depth + 1,
                )
                if nested_entries:
                    return nested_info, nested_entries
        return info, entries
    if etype.startswith("url") and info.get("url"):
        # Single nested URL without playlist wrapper — treat as non-subscribable
        return info, []
    return info, []


def _entry_video_url(entry: dict) -> Optional[str]:
    return entry.get("webpage_url") or entry.get("url")


def _entry_id(entry: dict) -> Optional[str]:
    eid = entry.get("id")
    if eid is not None:
        return str(eid)
    url = _entry_video_url(entry)
    return url


def _is_subscriber_only_entry(entry: dict) -> bool:
    """True when yt-dlp marks the entry as channel member-only (subscriber_only availability)."""
    return str(entry.get("availability") or "") == "subscriber_only"


def coerce_optional_bool(value: Any, *, default: bool = False, field_name: str = "value") -> bool:
    """Parse optional JSON booleans for subscription settings."""
    if value is None:
        return default
    try:
        return _coerce_bool(value)
    except ValueError as exc:
        raise ValueError(f"{field_name} must be a boolean") from exc


@dataclass
class SubscriptionInfo:
    id: str
    name: str
    url: str
    enabled: bool = True
    check_interval_minutes: int = 60
    download_type: str = "video"
    codec: str = "auto"
    format: str = "any"
    quality: str = "best"
    folder: str = ""
    custom_name_prefix: str = ""
    auto_start: bool = True
    playlist_item_limit: int = 0
    split_by_chapters: bool = False
    sponsorblock: bool = False
    chapter_template: str = ""
    subtitle_language: str = "en"
    subtitle_mode: str = "prefer_manual"
    ytdl_options_presets: list[str] = field(default_factory=list)
    ytdl_options_overrides: dict[str, Any] = field(default_factory=dict)
    title_regex: str = ""
    skip_subscriber_only: bool = False
    # A fixed range applied to every video the subscription downloads. Only
    # sensible for channels with a consistent format (a standing intro, a fixed
    # sponsor read); left unset, videos download whole. Old stored records
    # predate these fields and take the defaults — _from_stored filters by
    # field name, so nothing needs migrating.
    clip_start: Optional[float] = None
    clip_end: Optional[float] = None
    last_checked: Optional[float] = None
    seen_ids: list[str] = field(default_factory=list)
    error: Optional[str] = None
    timestamp: float = field(default_factory=time.time)

    def seen_set(self) -> set[str]:
        return set(self.seen_ids)

    def to_public_dict(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "url": self.url,
            "enabled": self.enabled,
            "check_interval_minutes": self.check_interval_minutes,
            "download_type": self.download_type,
            "codec": self.codec,
            "format": self.format,
            "quality": self.quality,
            "folder": self.folder,
            "title_regex": self.title_regex,
            "skip_subscriber_only": self.skip_subscriber_only,
            "clip_start": self.clip_start,
            "clip_end": self.clip_end,
            "last_checked": self.last_checked,
            "seen_count": len(self.seen_ids),
            "error": self.error,
        }


def _subscription_to_record(sub: SubscriptionInfo) -> dict[str, Any]:
    return {
        "id": sub.id,
        "name": sub.name,
        "url": sub.url,
        "enabled": sub.enabled,
        "check_interval_minutes": sub.check_interval_minutes,
        "download_type": sub.download_type,
        "codec": sub.codec,
        "format": sub.format,
        "quality": sub.quality,
        "folder": sub.folder,
        "custom_name_prefix": sub.custom_name_prefix,
        "auto_start": sub.auto_start,
        "playlist_item_limit": sub.playlist_item_limit,
        "split_by_chapters": sub.split_by_chapters,
        "sponsorblock": sub.sponsorblock,
        "chapter_template": sub.chapter_template,
        "subtitle_language": sub.subtitle_language,
        "subtitle_mode": sub.subtitle_mode,
        "ytdl_options_presets": list(sub.ytdl_options_presets),
        "ytdl_options_overrides": sub.ytdl_options_overrides,
        "title_regex": sub.title_regex,
        "skip_subscriber_only": sub.skip_subscriber_only,
        "clip_start": sub.clip_start,
        "clip_end": sub.clip_end,
        "last_checked": sub.last_checked,
        "seen_ids": list(sub.seen_ids),
        "error": sub.error,
    }


def _normalize_subscription_record(rec: dict[str, Any]) -> dict[str, Any]:
    """Migrate legacy ytdl_options_preset (str) to ytdl_options_presets (list)."""
    out = dict(rec)
    if "ytdl_options_presets" not in out:
        old = out.pop("ytdl_options_preset", None)
        if old is None:
            out["ytdl_options_presets"] = []
        elif isinstance(old, list):
            out["ytdl_options_presets"] = [str(x).strip() for x in old if str(x).strip()]
        elif isinstance(old, str):
            out["ytdl_options_presets"] = [old.strip()] if old.strip() else []
        else:
            out["ytdl_options_presets"] = []
    else:
        out.pop("ytdl_options_preset", None)
    return out


def _subscription_from_record(record: Any) -> Optional[SubscriptionInfo]:
    field_names = {f.name for f in fields(SubscriptionInfo)}
    if isinstance(record, SubscriptionInfo):
        return record
    if isinstance(record, dict):
        try:
            normalized = _normalize_subscription_record(dict(record))
            return SubscriptionInfo(**{k: v for k, v in normalized.items() if k in field_names})
        except TypeError:
            return None
    return None


def _normalize_title_regex_value(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    return str(value).strip()


def validate_title_regex(value: Any) -> str:
    """Return stored title regex string; non-empty values must compile (re.error on failure)."""
    s = _normalize_title_regex_value(value)
    if s:
        re.compile(s)
    return s


# The name is a display label the user picks; it is persisted and broadcast to
# every connected client, so keep it a bounded single-line string.
SUBSCRIPTION_NAME_MAX_LENGTH = 200


def validate_subscription_name(value: Any) -> str:
    """Return a stored subscription name, or raise ValueError if unusable."""
    if not isinstance(value, str):
        raise ValueError("name must be a string")
    # Collapse newlines/tabs so a pasted title can't break the table layout.
    name = " ".join(value.split())
    if not name:
        raise ValueError("name must not be empty")
    if len(name) > SUBSCRIPTION_NAME_MAX_LENGTH:
        raise ValueError(f"name must be at most {SUBSCRIPTION_NAME_MAX_LENGTH} characters")
    return name


def validate_subscription_folder(value: Any) -> str:
    """Return a stored subscription folder, or raise ValueError if unusable.

    The folder is relative to the configured download directory, and the
    authoritative check still happens at download time in ``DownloadQueue`` —
    that is where ``CUSTOM_DIRS``, ``CREATE_CUSTOM_DIRS`` and the
    resolves-inside-the-base-directory rule live, and where the directory is
    created. This rejects only values that could never be valid, so an edit is
    refused while the user is looking at it rather than silently failing every
    check from then on. An empty folder is valid and means the base directory.
    """
    if value is None:
        return ""
    if not isinstance(value, str):
        raise ValueError("folder must be a string")
    folder = value.strip()
    if not folder:
        return ""
    if os.path.isabs(folder):
        raise ValueError("folder must be relative to the download directory")
    # Check both separators: the value is stored as typed, and a Windows-style
    # path would otherwise carry an unexamined '..' past this point.
    if any(part == ".." for part in folder.replace("\\", "/").split("/")):
        raise ValueError('folder must not contain ".."')
    return folder


def _coerce_bool(value: Any) -> bool:
    """Accept JSON booleans and common string forms used by API clients."""
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"true", "1", "on"}:
            return True
        if lowered in {"false", "0", "off"}:
            return False
    raise ValueError("enabled must be a boolean")


class SubscriptionNotifier:
    """Hook for Socket.IO / UI updates."""

    async def subscription_added(self, sub: SubscriptionInfo) -> None:
        raise NotImplementedError

    async def subscription_updated(self, sub: SubscriptionInfo) -> None:
        raise NotImplementedError

    async def subscription_removed(self, sub_id: str) -> None:
        raise NotImplementedError

    async def subscriptions_all(self, subs: list[SubscriptionInfo]) -> None:
        raise NotImplementedError


class SubscriptionManager:
    def __init__(self, config, download_queue, notifier: SubscriptionNotifier):
        self.config = config
        self.dqueue = download_queue
        self.notifier = notifier
        pdir = config.STATE_DIR
        if not os.path.isdir(pdir):
            os.makedirs(pdir, exist_ok=True)
        self._legacy_path = os.path.join(pdir, "subscriptions")
        self._path = os.path.join(pdir, "subscriptions.json")
        self._store = AtomicJsonStore(self._path, kind="subscriptions")
        self._subs: dict[str, SubscriptionInfo] = {}
        self._url_index: dict[str, str] = {}  # normalized url -> id
        self._pending_urls: set[str] = set()
        self._checks_in_flight: set[str] = set()  # subscription ids being checked right now
        self._lock = asyncio.Lock()
        self._loop_task: Optional[asyncio.Task] = None
        self._load_all()

    def close(self) -> None:
        # No persistent shelf handle to close.
        return

    def _normalize_url(self, url: str) -> str:
        return (url or "").strip()

    def _normalize_seen_ids(self, seen_ids: list[str]) -> list[str]:
        max_seen = int(getattr(self.config, "SUBSCRIPTION_MAX_SEEN_IDS", 50000))
        normalized = [str(sid) for sid in dict.fromkeys(seen_ids)]
        if len(normalized) > max_seen:
            normalized = normalized[:max_seen]
        return normalized

    def _load_all(self) -> None:
        payload = self._store.load()
        loaded_from_legacy = False
        if payload is not None:
            records = payload.get("items") or []
        else:
            legacy_items = read_legacy_shelf(self._legacy_path)
            records = [raw for _key, raw in legacy_items] if legacy_items else []
            if records:
                loaded_from_legacy = True

        loaded_subs = self._iter_valid_subs(records)
        compact_records = []
        for sub in loaded_subs:
            sub.seen_ids = self._normalize_seen_ids(sub.seen_ids)
            self._subs[sub.id] = sub
            self._url_index[self._normalize_url(sub.url)] = sub.id
            compact_records.append(_subscription_to_record(sub))

        if loaded_from_legacy or (
            payload is not None
            and (
                payload.get("schema_version") != self._store.schema_version
                or compact_records != records
            )
        ):
            self._store.save({"items": compact_records})

    def _iter_valid_subs(self, records: list[Any]) -> list[SubscriptionInfo]:
        subs: list[SubscriptionInfo] = []
        for record in records:
            sub = _subscription_from_record(record)
            if sub is not None:
                subs.append(sub)
        return subs

    def _save_locked(self) -> None:
        self._store.save({"items": [_subscription_to_record(sub) for sub in self._subs.values()]})

    def _scan_extra_opts(
        self,
        ytdl_options_presets: Optional[list[str]],
        ytdl_options_overrides: Optional[dict[str, Any]],
    ) -> dict[str, Any]:
        """Merge configured presets (in order) with per-subscription overrides.

        Applied on top of the global YTDL_OPTIONS when scanning a
        subscription's feed, so cookies/impersonation/etc. configured via a
        preset or override also take effect during the flat-playlist scan,
        not just the eventual per-video download. (The global YTDL_OPTIONS base
        is already spread into the scan params by ``_build_ydl_params``.)
        """
        return merge_ytdl_option_layers(
            ytdl_options_presets, ytdl_options_overrides, self.config.YTDL_OPTIONS_PRESETS
        )

    async def _queue_subscription_entries(
        self,
        entries: list[dict],
        *,
        download_type: str,
        codec: str,
        format: str,
        quality: str,
        folder: str,
        custom_name_prefix: str,
        playlist_item_limit: int,
        auto_start: bool,
        split_by_chapters: bool,
        chapter_template: str,
        subtitle_language: str,
        subtitle_mode: str,
        ytdl_options_presets: Optional[list[str]] = None,
        ytdl_options_overrides: Optional[dict[str, Any]] = None,
        clip_start: Optional[float] = None,
        clip_end: Optional[float] = None,
        sponsorblock: bool = False,
    ) -> tuple[list[str], list[str]]:
        queued_ids: list[str] = []
        queue_errors: list[str] = []
        presets = list(ytdl_options_presets or [])
        for ent in entries:
            eid = _entry_id(ent)
            vurl = _entry_video_url(ent)
            if not eid or not vurl:
                continue
            queue_entry = dict(ent)
            if "id" not in queue_entry:
                queue_entry["id"] = eid
            queue_entry["_type"] = "video"
            queue_entry["webpage_url"] = vurl
            result = await self.dqueue.add_entry(
                queue_entry,
                download_type,
                codec,
                format,
                quality,
                folder or None,
                custom_name_prefix,
                playlist_item_limit,
                auto_start,
                split_by_chapters,
                chapter_template or None,
                subtitle_language,
                subtitle_mode,
                presets,
                ytdl_options_overrides,
                clip_start,
                clip_end,
                sponsorblock=sponsorblock,
            )
            if isinstance(result, dict) and result.get("status") == "error":
                msg = str(result.get("msg") or f"Queueing failed for {vurl}")
                queue_errors.append(msg)
                log.warning("Subscription queueing failed for %s: %s", vurl, msg)
                continue
            queued_ids.append(eid)
        return queued_ids, queue_errors

    def list_all(self) -> list[SubscriptionInfo]:
        return list(self._subs.values())

    def get(self, sub_id: str) -> Optional[SubscriptionInfo]:
        return self._subs.get(sub_id)

    def start_background_loop(self) -> None:
        if self._loop_task is not None and not self._loop_task.done():
            return
        # bg_tasks.create_task already logs unexpected task failures with the name.
        self._loop_task = bg_tasks.create_task(self._periodic_loop(), name="subscription_loop")

    async def _periodic_loop(self) -> None:
        while True:
            await asyncio.sleep(60)
            try:
                await self.run_due_checks()
            except Exception as e:
                log.exception("Subscription periodic check error: %s", e)

    async def run_due_checks(self) -> None:
        now = time.time()
        due: list[SubscriptionInfo] = []
        async with self._lock:
            for sub in list(self._subs.values()):
                if not sub.enabled:
                    continue
                interval_sec = max(60, int(sub.check_interval_minutes) * 60)
                if sub.last_checked is None:
                    due.append(sub)
                    continue
                if now - sub.last_checked < interval_sec:
                    continue
                due.append(sub)
        await self._check_many(due)

    async def _check_many(self, subs: list[SubscriptionInfo]) -> None:
        """Check subscriptions with bounded concurrency so one slow feed does
        not serialize the rest. Failures are isolated per subscription."""
        if not subs:
            return
        sem = asyncio.Semaphore(_MAX_CONCURRENT_CHECKS)

        async def _guarded(sub: SubscriptionInfo) -> None:
            async with sem:
                await self._check_one_unlocked(sub)

        results = await asyncio.gather(
            *(_guarded(sub) for sub in subs), return_exceptions=True
        )
        for sub, result in zip(subs, results):
            if isinstance(result, Exception):
                log.error(
                    "Subscription check crashed for %s: %s",
                    sub.name,
                    result,
                    exc_info=result,
                )

    async def add_subscription(
        self,
        url: str,
        *,
        check_interval_minutes: int,
        download_type: str,
        codec: str,
        format: str,
        quality: str,
        folder: str,
        custom_name_prefix: str,
        auto_start: bool,
        playlist_item_limit: int,
        split_by_chapters: bool,
        chapter_template: str,
        subtitle_language: str,
        subtitle_mode: str,
        ytdl_options_presets: Optional[list[str]] = None,
        ytdl_options_overrides: Optional[dict[str, Any]] = None,
        sponsorblock: bool = False,
        title_regex: Any = None,
        skip_subscriber_only: Any = None,
        clip_start: Optional[float] = None,
        clip_end: Optional[float] = None,
    ) -> dict:
        url = self._normalize_url(url)
        if not url:
            return {"status": "error", "msg": "Missing URL"}
        # SSRF guard: block non-http(s) schemes and internal/metadata hosts
        # before yt-dlp fetches the feed. May do a DNS lookup, so run off-loop.
        # The scan's own options pick the proxy, so a feed fetched through one
        # is not resolved here — see validate_url.
        proxies = download_proxies({
            **self.config.YTDL_OPTIONS,
            **self._scan_extra_opts(ytdl_options_presets, ytdl_options_overrides),
        })
        url_error = await asyncio.get_running_loop().run_in_executor(
            None, partial(validate_url, url,
                          allow_private=getattr(self.config, "ALLOW_PRIVATE_ADDRESSES", False),
                          proxies=proxies))
        if url_error is not None:
            log.warning('Rejected subscription URL "%s": %s', url, url_error)
            return {"status": "error", "msg": url_error}
        try:
            title_regex_stored = validate_title_regex(title_regex)
        except re.error as exc:
            return {"status": "error", "msg": f"Invalid title_regex: {exc}"}
        try:
            skip_so = coerce_optional_bool(
                skip_subscriber_only,
                default=False,
                field_name="skip_subscriber_only",
            )
        except ValueError as exc:
            return {"status": "error", "msg": str(exc)}

        async with self._lock:
            if url in self._url_index or url in self._pending_urls:
                return {"status": "error", "msg": "This URL is already subscribed"}
            self._pending_urls.add(url)

        try:
            scan_first = max(int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50)), 1)
            scan_extra_opts = self._scan_extra_opts(ytdl_options_presets, ytdl_options_overrides)
            try:
                info, entries = await asyncio.get_running_loop().run_in_executor(
                    None,
                    partial(extract_flat_playlist, self.config, url, scan_first, extra_opts=scan_extra_opts),
                )
            except yt_dlp.utils.YoutubeDLError as exc:
                return {"status": "error", "msg": str(exc)}

            if not info:
                return {"status": "error", "msg": "Could not resolve URL"}

            etype = info.get("_type") or "video"
            if etype not in ("playlist", "channel"):
                return {"status": "error", "msg": VIDEO_ONLY_MSG}

            name = (
                info.get("title")
                or info.get("channel")
                or info.get("playlist_title")
                or info.get("uploader")
                or url
            )

            seen_entries = [ent for ent in entries if _is_media_entry(ent)]
            all_ids: list[str] = []
            for ent in seen_entries:
                if ent.get("live_status") == "is_upcoming":
                    continue  # Don't mark scheduled streams as seen; queue them when they go live
                eid = _entry_id(ent)
                if eid:
                    all_ids.append(eid)

            sub = SubscriptionInfo(
                id=str(uuid.uuid4()),
                name=str(name),
                url=url,
                enabled=True,
                check_interval_minutes=max(1, int(check_interval_minutes)),
                download_type=download_type,
                codec=codec,
                format=format,
                quality=quality,
                folder=folder or "",
                custom_name_prefix=custom_name_prefix or "",
                auto_start=bool(auto_start),
                playlist_item_limit=int(playlist_item_limit),
                split_by_chapters=bool(split_by_chapters),
                sponsorblock=bool(sponsorblock),
                chapter_template=chapter_template or "",
                subtitle_language=subtitle_language,
                subtitle_mode=subtitle_mode,
                ytdl_options_presets=list(ytdl_options_presets or []),
                ytdl_options_overrides=dict(ytdl_options_overrides or {}),
                title_regex=title_regex_stored,
                skip_subscriber_only=skip_so,
                clip_start=clip_start,
                clip_end=clip_end,
                last_checked=time.time(),
                seen_ids=list(dict.fromkeys(all_ids)),
                error=None,
            )

            async with self._lock:
                if url in self._url_index:
                    return {"status": "error", "msg": "This URL is already subscribed"}
                self._subs[sub.id] = sub
                self._url_index[url] = sub.id
                try:
                    self._save_locked()
                except Exception:
                    self._subs.pop(sub.id, None)
                    self._url_index.pop(url, None)
                    raise

            await self.notifier.subscription_added(sub)
            return {"status": "ok", "subscription": sub.to_public_dict()}
        finally:
            async with self._lock:
                self._pending_urls.discard(url)

    async def delete_subscriptions(self, ids: list[str]) -> dict:
        removed: list[str] = []
        async with self._lock:
            previous_subs = self._subs.copy()
            previous_index = self._url_index.copy()
            for sid in ids:
                sub = self._subs.pop(sid, None)
                if sub:
                    normalized_url = self._normalize_url(sub.url)
                    self._url_index.pop(normalized_url, None)
                    removed.append(sid)
            if removed:
                try:
                    self._save_locked()
                except Exception:
                    self._subs = previous_subs
                    self._url_index = previous_index
                    raise
        for sid in removed:
            await self.notifier.subscription_removed(sid)
        return {"status": "ok"}

    async def update_subscription(self, sub_id: str, changes: dict) -> dict:
        validated_name: Optional[str] = None
        if "name" in changes:
            try:
                validated_name = validate_subscription_name(changes["name"])
            except ValueError as exc:
                return {"status": "error", "msg": str(exc)}

        validated_folder: Optional[str] = None
        if "folder" in changes:
            try:
                validated_folder = validate_subscription_folder(changes["folder"])
            except ValueError as exc:
                return {"status": "error", "msg": str(exc)}

        validated_tr: Optional[str] = None
        if "title_regex" in changes:
            try:
                validated_tr = validate_title_regex(changes["title_regex"])
            except re.error as exc:
                return {"status": "error", "msg": f"Invalid title_regex: {exc}"}

        skip_so_set = False
        validated_skip_so = False
        if "skip_subscriber_only" in changes:
            try:
                validated_skip_so = coerce_optional_bool(
                    changes["skip_subscriber_only"],
                    field_name="skip_subscriber_only",
                )
                skip_so_set = True
            except ValueError as exc:
                return {"status": "error", "msg": str(exc)}

        enabled_set = False
        validated_enabled = False
        if "enabled" in changes:
            try:
                validated_enabled = _coerce_bool(changes["enabled"])
                enabled_set = True
            except ValueError as exc:
                return {"status": "error", "msg": str(exc)}

        interval_set = False
        validated_interval = 0
        if "check_interval_minutes" in changes:
            try:
                validated_interval = max(1, int(changes["check_interval_minutes"]))
            except (TypeError, ValueError):
                return {"status": "error", "msg": "check_interval_minutes must be an integer"}
            interval_set = True

        async with self._lock:
            sub = self._subs.get(sub_id)
            if not sub:
                return {"status": "error", "msg": "Subscription not found"}
            previous = copy.deepcopy(sub)
            old_enabled = sub.enabled

            if enabled_set:
                sub.enabled = validated_enabled
            if interval_set:
                sub.check_interval_minutes = validated_interval
            if validated_name is not None:
                sub.name = validated_name
            if validated_folder is not None:
                # Applies to future downloads only; files already downloaded
                # stay where they are.
                sub.folder = validated_folder
            if validated_tr is not None:
                sub.title_regex = validated_tr
            if skip_so_set:
                sub.skip_subscriber_only = validated_skip_so

            try:
                self._save_locked()
            except Exception:
                self._subs[sub_id] = previous
                raise
            updated = sub
        if "enabled" in changes and updated.enabled != old_enabled:
            log.info(
                "Subscription %s %s",
                updated.name,
                "resumed" if updated.enabled else "paused",
            )
        await self.notifier.subscription_updated(updated)
        return {"status": "ok", "subscription": updated.to_public_dict()}

    async def check_now(self, ids: Optional[list[str]] = None) -> dict:
        async with self._lock:
            targets = (
                [self._subs[i] for i in ids if i in self._subs]
                if ids
                else [s for s in self._subs.values() if s.enabled]
            )
        log.info(
            "Manual subscription check requested for %d subscription(s)",
            len(targets),
        )
        await self._check_many(targets)
        return {"status": "ok"}

    async def _check_one_unlocked(self, sub: SubscriptionInfo) -> None:
        sid = sub.id
        # Prevent overlapping checks for the same subscription (e.g. the periodic
        # loop and a manual check-now firing together), which could double-queue
        # entries and drop seen_ids via a read-modify-write race.
        async with self._lock:
            if sid in self._checks_in_flight:
                log.info("Subscription check already in progress for %s, skipping", sub.name)
                return
            self._checks_in_flight.add(sid)
        try:
            await self._check_one_inner(sub)
        finally:
            async with self._lock:
                self._checks_in_flight.discard(sid)

    async def _check_one_inner(self, sub: SubscriptionInfo) -> None:
        sid = sub.id
        scan = int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50))
        # ytdl_options_presets/overrides are set at subscription creation and
        # never mutated afterwards (update_subscription doesn't allow it), so
        # reading them off `sub` here without holding the lock is safe.
        scan_extra_opts = self._scan_extra_opts(sub.ytdl_options_presets, sub.ytdl_options_overrides)
        log.info("Checking subscription: %s", sub.name)
        try:
            info, entries = await asyncio.get_running_loop().run_in_executor(
                None,
                partial(extract_flat_playlist, self.config, sub.url, scan, extra_opts=scan_extra_opts),
            )
        except yt_dlp.utils.YoutubeDLError as exc:
            async with self._lock:
                cur = self._subs.get(sid)
                if cur:
                    previous = copy.deepcopy(cur)
                    cur.error = str(exc)
                    cur.last_checked = time.time()
                    try:
                        self._save_locked()
                    except Exception:
                        self._subs[sid] = previous
                        raise
                    sub = cur
            log.warning("Subscription check failed for %s: %s", sub.name, exc)
            await self.notifier.subscription_updated(sub)
            return
        entries = [ent for ent in entries if _is_media_entry(ent)]

        etype = (info or {}).get("_type") or "video"
        if etype == "video":
            async with self._lock:
                cur = self._subs.get(sid)
                if cur:
                    previous = copy.deepcopy(cur)
                    cur.error = VIDEO_ONLY_MSG
                    cur.last_checked = time.time()
                    try:
                        self._save_locked()
                    except Exception:
                        self._subs[sid] = previous
                        raise
                    sub = cur
            log.warning("Subscription %s no longer resolves to a subscribable feed", sub.name)
            await self.notifier.subscription_updated(sub)
            return
        if not entries:
            async with self._lock:
                cur = self._subs.get(sid)
                if cur:
                    previous = copy.deepcopy(cur)
                    cur.last_checked = time.time()
                    cur.error = None
                    try:
                        self._save_locked()
                    except Exception:
                        self._subs[sid] = previous
                        raise
                    sub = cur
            log.warning("Subscription check finished for %s: No entries found", sub.name)
            await self.notifier.subscription_updated(sub)
            return

        async with self._lock:
            cur = self._subs.get(sid)
            if not cur:
                return
            seen = cur.seen_set()
            seen_ids_snapshot = list(cur.seen_ids)
            dl_type = cur.download_type
            dl_codec = cur.codec
            dl_format = cur.format
            dl_quality = cur.quality
            dl_folder = cur.folder
            dl_prefix = cur.custom_name_prefix
            dl_plimit = cur.playlist_item_limit
            dl_autostart = cur.auto_start
            dl_split = cur.split_by_chapters
            dl_sponsorblock = cur.sponsorblock
            dl_chapter = cur.chapter_template
            dl_sublang = cur.subtitle_language
            dl_submode = cur.subtitle_mode
            dl_ytdl_presets = list(cur.ytdl_options_presets)
            dl_ytdl_overrides = dict(cur.ytdl_options_overrides)
            dl_title_regex = cur.title_regex or ""
            dl_skip_subscriber_only = bool(cur.skip_subscriber_only)
            dl_clip_start = cur.clip_start
            dl_clip_end = cur.clip_end

        new_entries: list[dict] = []
        for ent in entries:
            eid = _entry_id(ent)
            if not eid:
                continue
            # Seen entries that are currently live are deliberately re-queued:
            # a stream first seen as 'upcoming' must still be captured once it
            # goes live. The download queue dedups by URL while a capture is
            # in flight, so this can't double-queue an active capture.
            if eid in seen and ent.get("live_status") != "is_live":
                continue
            new_entries.append(ent)

        pattern_re: Optional[re.Pattern[str]] = None
        if dl_title_regex:
            try:
                pattern_re = re.compile(dl_title_regex)
            except re.error:
                log.warning(
                    "Invalid stored title_regex on subscription %s, ignoring filter",
                    sub.name,
                )

        queue_entries: list[dict] = []
        filtered_ids: list[str] = []
        for ent in new_entries:
            eid = _entry_id(ent)
            if pattern_re is not None:
                title = str(ent.get("title") or "")
                if not pattern_re.search(title):
                    if eid:
                        filtered_ids.append(eid)
                    continue
            queue_entries.append(ent)

        subscriber_filtered_ids: list[str] = []
        if dl_skip_subscriber_only:
            kept_entries: list[dict] = []
            for ent in queue_entries:
                eid = _entry_id(ent)
                if _is_subscriber_only_entry(ent):
                    if eid:
                        subscriber_filtered_ids.append(eid)
                    continue
                kept_entries.append(ent)
            queue_entries = kept_entries

        queued_ids, queue_errors = await self._queue_subscription_entries(
            queue_entries,
            download_type=dl_type,
            codec=dl_codec,
            format=dl_format,
            quality=dl_quality,
            folder=dl_folder,
            custom_name_prefix=dl_prefix,
            playlist_item_limit=dl_plimit,
            auto_start=dl_autostart,
            split_by_chapters=dl_split,
            sponsorblock=dl_sponsorblock,
            chapter_template=dl_chapter or "",
            subtitle_language=dl_sublang,
            subtitle_mode=dl_submode,
            ytdl_options_presets=dl_ytdl_presets,
            ytdl_options_overrides=dl_ytdl_overrides,
            clip_start=dl_clip_start,
            clip_end=dl_clip_end,
        )
        log.info(
            "Subscription check finished for %s: %d new, %d filtered, %d subscriber_skipped, %d queued, %d failed",
            sub.name,
            len(new_entries),
            len(filtered_ids),
            len(subscriber_filtered_ids),
            len(queued_ids),
            len(queue_errors),
        )

        merged = list(
            dict.fromkeys(
                queued_ids + filtered_ids + subscriber_filtered_ids + seen_ids_snapshot
            )
        )
        max_seen = int(getattr(self.config, "SUBSCRIPTION_MAX_SEEN_IDS", 50000))
        if len(merged) > max_seen:
            merged = merged[:max_seen]

        async with self._lock:
            cur = self._subs.get(sid)
            if not cur:
                return
            previous = copy.deepcopy(cur)
            cur.seen_ids = merged
            cur.last_checked = time.time()
            cur.error = "; ".join(queue_errors[:3]) if queue_errors else None
            try:
                self._save_locked()
            except Exception:
                self._subs[sid] = previous
                raise
            sub = cur
        await self.notifier.subscription_updated(sub)

    async def emit_all(self) -> None:
        await self.notifier.subscriptions_all(self.list_all())