app.py

"""
Advanced Stream Bot — v3.3.0
- Fixed: frames_encoded now counts decoded input frames (encoder buffering no longer causes 0)
- Fixed: live_log_lines_user cleared on stream start so stale logs don't persist
- Fixed: last_sent reset on stream start so bubble refreshes immediately
- Fixed: raw FFmpeg logs via queue-drained thread (bitrate, fps, codec init, errors)
- Fixed: log tail display increased 8→15 lines
- New: GET /poll/{chat_id} endpoint — returns pending live-view edit for proactive refresh
- Live view: status+log bubble auto-edits every 2s (logg.py pattern)
- Pause/Resume/Refresh/Close live view controls
- Force Reboot button + /reboot command — last-resort recovery
- Fixed: error field no longer shows raw log lines
- Fixed: error cleared on clean stop/abort
- Fixed: stats (uptime/frames/bytes) update correctly in live view
- Fixed: abort works instantly during reconnect sleep
- Fixed: watchdog does not mark error after intentional stop
- Fixed: stale edits never pile up in outbound queue (cap=1 per message)
- SSE /events/{chat_id} for real-time web dashboard
- Scheduler-triggered streams queue outbound notifications
"""

import logging
import threading
import time
import datetime
import traceback
import fractions
import json
import os
import re
import asyncio
from urllib.parse import urlparse
from pathlib import Path
import io

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import queue
import av
from PIL import Image, ImageEnhance, UnidentifiedImageError
from apscheduler.schedulers.background import BackgroundScheduler

# ──────────────────────────────────────────────
# VERSION & APP
# ──────────────────────────────────────────────
APP_VERSION = "3.3.0"

app = FastAPI(title="Advanced Stream Bot", version=APP_VERSION)
scheduler = BackgroundScheduler(timezone="UTC")
# Per-job metadata store — APScheduler Job objects are frozen;
# we cannot attach arbitrary attrs to them. Store meta here instead.
_job_store: dict = {}  # job_id -> {"output_url": ..., "input_url": ...}


# ──────────────────────────────────────────────
# LOGGING
# ──────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(threadName)s %(module)s:%(lineno)d %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("stream_bot")

live_log_lines_global = []
MAX_GLOBAL_LOG_LINES = 200


def append_global_live_log(line: str):
    live_log_lines_global.append(line)
    if len(live_log_lines_global) > MAX_GLOBAL_LOG_LINES:
        live_log_lines_global.pop(0)


class GlobalListHandler(logging.Handler):
    def emit(self, record):
        append_global_live_log(self.format(record))


_gh = GlobalListHandler()
_gh.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
logger.addHandler(_gh)

# ──────────────────────────────────────────────
# VALIDATION CONSTANTS
# ──────────────────────────────────────────────
SUPPORTED_VIDEO_CODECS = ["libx264", "h264_nvenc", "h264_qsv", "libx265", "hevc_nvenc", "hevc_qsv", "copy"]
SUPPORTED_AUDIO_CODECS = ["aac", "opus", "libmp3lame", "copy"]
LOGO_POSITIONS = ["top_left", "top_right", "bottom_left", "bottom_right", "center"]
FFMPEG_PRESETS = ["ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow", "slower", "veryslow"]
OUTPUT_FORMATS = ["flv", "mpegts", "mp4", "hls"]

# Common resolution presets for the picker
RESOLUTION_PRESETS = [
    ("360p",  "640x360"),
    ("480p",  "854x480"),
    ("720p",  "1280x720"),
    ("1080p", "1920x1080"),
    ("1440p", "2560x1440"),
    ("4K",    "3840x2160"),
    ("Source","source"),
]

QUALITY_PRESETS = {
    "low":    {"video_bitrate": "800k",  "audio_bitrate": "96k",  "ffmpeg_preset": "superfast", "resolution": "854x480",  "fps": 30},
    "medium": {"video_bitrate": "1500k", "audio_bitrate": "128k", "ffmpeg_preset": "medium",    "resolution": "1280x720", "fps": 30},
    "high":   {"video_bitrate": "3000k", "audio_bitrate": "192k", "ffmpeg_preset": "fast",      "resolution": "1920x1080","fps": 30},
    "ultra":  {"video_bitrate": "6000k", "audio_bitrate": "256k", "ffmpeg_preset": "slow",      "resolution": "1920x1080","fps": 60},
    "source": {},
}

# ──────────────────────────────────────────────
# DEFAULT SETTINGS — every parameter exposed
# ──────────────────────────────────────────────
DEFAULT_USER_SETTINGS = {
    # Playlist / IO
    "input_url_playlist": [],
    "current_playlist_index": 0,
    "output_url": "rtmp://a.rtmp.youtube.com/live2/YOUR_KEY",
    "output_format": "flv",

    # Quality shortcut
    "quality_preset": "medium",

    # Video
    "video_codec": "libx264",
    "resolution": "1280x720",
    "fps": 30,
    "gop_size": 60,
    "video_bitrate": "1500k",
    "ffmpeg_preset": "medium",
    "video_pix_fmt": "yuv420p",
    "video_thread_count": 0,        # 0 = auto

    # Audio
    "audio_codec": "aac",
    "audio_bitrate": "128k",
    "audio_sample_rate": 44100,     # 0 = source
    "audio_channels": 2,            # 0 = source

    # Loop / playlist
    "loop_count": 0,                # 0=once, -1=infinite, N=N times
    "stop_on_error_in_playlist": True,
    "reconnect_on_stream_error": True,
    "reconnect_delay_seconds": 5,
    "max_reconnect_attempts": 3,

    # Connection / network
    "open_timeout_seconds": 15,
    "read_timeout_seconds": 30,

    # Logo
    "logo_enabled": False,
    "logo_data_bytes": None,
    "logo_mime_type": None,
    "logo_original_filename": None,
    "logo_position": "top_right",
    "logo_scale": 0.10,
    "logo_opacity": 0.85,
    "logo_margin_px": 10,

    # Logging
    "verbose_ffmpeg_log": False,    # show raw FFmpeg/libav lines in live log

    # Conversation state
    "current_step": None,
    "current_step_index": 0,
    "conversation_fields_list": [],
    "settings_editing_field": None,

    # UX mode: "send" = always new message (default), "edit" = edit-in-place
    "ux_mode": "send",
}

DEFAULT_SESSION_RUNTIME_STATE = {
    "streaming_state": "idle",      # idle|starting|streaming|paused|stopping|stopped|completed|error|reconnecting
    "stream_start_time": None,
    "frames_encoded": 0,
    "bytes_sent": 0,
    "stream_thread_ref": None,
    "watchdog_thread_ref": None,
    "pyav_objects": {
        "input_container": None,
        "output_container": None,
        "video_out_stream": None,
        "audio_out_stream": None,
        "logo_image_pil": None,
    },
    "live_log_lines_user": [],
    "error_notification_user": "",
    "stop_gracefully_flag": False,
    "current_loop_iteration": 0,
    "reconnect_attempt": 0,
    "last_frame_time": None,        # for watchdog
    "last_notified_state": None,    # for state-change notifications
    "status_message_id": None,      # message to auto-update in real-time
    "status_chat_id": None,
    "active_output_url": None,       # the URL actually being streamed to right now
}

# ──────────────────────────────────────────────
# SESSION MANAGEMENT
# ──────────────────────────────────────────────
user_sessions: dict = {}
session_locks: dict = {}
# Store the main event loop for cross-thread calls
_main_event_loop: asyncio.AbstractEventLoop = None

# ──────────────────────────────────────────────
# REAL-TIME EVENT QUEUE SYSTEM
# Each chat_id has a list of asyncio.Queue objects (one per SSE subscriber).
# When any state/log change happens, we push an event to all subscribers.
# ──────────────────────────────────────────────
_sse_subscribers: dict = {}   # chat_id -> list[asyncio.Queue]
_sse_lock = threading.Lock()

# Outbound message queue: background threads can enqueue messages here;
# they will be returned on the next /webhook response for that chat.
_outbound_message_queue: dict = {}   # chat_id -> list[dict]
_outbound_queue_lock = threading.Lock()


def _get_sse_subscribers(chat_id: int) -> list:
    with _sse_lock:
        return list(_sse_subscribers.get(chat_id, []))


def _register_sse_subscriber(chat_id: int, q: asyncio.Queue):
    with _sse_lock:
        _sse_subscribers.setdefault(chat_id, []).append(q)


def _unregister_sse_subscriber(chat_id: int, q: asyncio.Queue):
    with _sse_lock:
        lst = _sse_subscribers.get(chat_id, [])
        try:
            lst.remove(q)
        except ValueError:
            pass


def push_sse_event(chat_id: int, event: dict):
    """Push a JSON event to all SSE subscribers for this chat_id (thread-safe)."""
    subscribers = _get_sse_subscribers(chat_id)
    if not subscribers:
        return
    loop = _main_event_loop
    if not loop or not loop.is_running():
        return
    payload = json.dumps(event)
    for q in subscribers:
        try:
            loop.call_soon_threadsafe(q.put_nowait, payload)
        except Exception:
            pass


def enqueue_outbound_message(chat_id: int, msg: dict):
    """Queue a message dict to be returned on the next webhook response for chat_id."""
    with _outbound_queue_lock:
        _outbound_message_queue.setdefault(chat_id, []).append(msg)


def pop_outbound_messages(chat_id: int) -> list:
    """Pop and return all queued outbound messages for this chat_id."""
    with _outbound_queue_lock:
        msgs = _outbound_message_queue.pop(chat_id, [])
    return msgs


def get_user_session(chat_id: int) -> dict:
    if chat_id not in user_sessions:
        session_locks[chat_id] = threading.Lock()
        with session_locks[chat_id]:
            session = {}
            for k, v in DEFAULT_USER_SETTINGS.items():
                session[k] = list(v) if isinstance(v, list) else (dict(v) if isinstance(v, dict) else v)
            for k, v in DEFAULT_SESSION_RUNTIME_STATE.items():
                session[k] = dict(v) if isinstance(v, dict) else (list(v) if isinstance(v, list) else v)
            user_sessions[chat_id] = session
            logger.info(f"[Chat {chat_id}] New session created (in-memory).")
    return user_sessions[chat_id]


def reset_session_settings(chat_id: int):
    """Restore all settings to defaults, keep runtime state."""
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    with lock:
        for k, v in DEFAULT_USER_SETTINGS.items():
            session[k] = list(v) if isinstance(v, list) else (dict(v) if isinstance(v, dict) else v)
    append_user_live_log(chat_id, "Settings restored to defaults.")


def append_user_live_log(chat_id: int, line: str):
    session = get_user_session(chat_id)
    lock = session_locks.get(chat_id)
    entry = f"{datetime.datetime.now().strftime('%H:%M:%S')} {line}"
    if lock:
        with lock:
            session['live_log_lines_user'].append(entry)
            if len(session['live_log_lines_user']) > 100:
                session['live_log_lines_user'].pop(0)
    else:
        session['live_log_lines_user'].append(entry)
    # Push real-time log event to SSE subscribers
    push_sse_event(chat_id, {"type": "log", "line": entry})


# ──────────────────────────────────────────────
# TELEGRAM API HELPERS
# (Webhook-only mode: no outbound Telegram API calls)
# ──────────────────────────────────────────────
def push_message(chat_id: int, text: str, reply_markup=None, parse_mode="HTML"):
    """
    In webhook-only mode outbound calls are not possible.
    Log the notification so it appears in /logs instead.
    """
    import html
    plain = html.unescape(re.sub(r'<[^>]+>', '', text))
    logger.info(f"[Chat {chat_id}] [push_message suppressed] {plain[:200]}")
    append_user_live_log(chat_id, f"[notification] {plain[:200]}")


# ──────────────────────────────────────────────
# TELEGRAM RESPONSE BUILDERS
# ──────────────────────────────────────────────
def send_message(chat_id: int, text: str, parse_mode="HTML", reply_markup=None):
    """Build a sendMessage response dict (for webhook response)."""
    resp = {"method": "sendMessage", "chat_id": chat_id, "text": text, "parse_mode": parse_mode}
    if reply_markup:
        resp["reply_markup"] = reply_markup
    return resp


def edit_message_text(chat_id: int, message_id: int, text: str, parse_mode="HTML", reply_markup=None):
    resp = {"method": "editMessageText", "chat_id": chat_id, "message_id": message_id,
            "text": text, "parse_mode": parse_mode}
    if reply_markup:
        resp["reply_markup"] = reply_markup
    return resp


def answer_callback_query(cq_id: str, text: str = None, show_alert: bool = False):
    resp = {"method": "answerCallbackQuery", "callback_query_id": cq_id}
    if text:
        resp["text"] = text
    if show_alert:
        resp["show_alert"] = True
    return resp


# ──────────────────────────────────────────────
# LIVE VIEW SYSTEM  (inspired by logg.py)
# ──────────────────────────────────────────────
# Pattern taken from logg.py:
#   • Each chat has a "live_view" entry: message_id + last_sent_text + show flag
#   • A background thread every UPDATE_INTERVAL seconds builds the fresh combined
#     status+log text and queues an editMessageText for that message_id.
#   • On next webhook response the edit is piggybacked — the bubble updates in place.
#   • Pause/Resume toggle show_live_view, suppressing/resuming edits (same as logg.py).
#   • Refresh forces an immediate rebuild+enqueue regardless of interval.
#   • Auto-registers when user taps 📊 Status or 📋 Logs.
#   • Auto-unregisters when stream reaches a terminal state.
#   • SSE events are pushed alongside every edit for web dashboard clients.
#
# UPDATE_INTERVAL: how often the background thread re-edits the live bubble (seconds)
LIVE_VIEW_UPDATE_INTERVAL = 2

_live_views: dict = {}          # chat_id -> {message_id, last_sent, show, paused_at, mode}
_live_view_lock = threading.Lock()


def lock_for(chat_id: int):
    return session_locks.get(chat_id, threading.Lock())


def _build_live_view_text(chat_id: int, mode: str = "status") -> str:
    """
    Build the combined live-view message text.
    mode="status" → status block + last 8 log lines
    mode="logs"   → last 30 log lines only (like logg.py /logs)
    """
    session = get_user_session(chat_id)
    state = session.get('streaming_state', 'idle')

    with _live_view_lock:
        info = _live_views.get(chat_id, {})
        is_paused = not info.get("show", True)

    if mode == "logs":
        logs = session.get('live_log_lines_user', [])
        log_tail = "\n".join(logs[-30:]) if logs else "No logs yet."
        pause_note = "\n\n<i>⏸ Updates paused</i>" if is_paused else ""
        return (
            f"📋 <b>Live Logs</b> {'(paused)' if is_paused else '(updating…)'}\n"
            f"<pre>{esc(log_tail)}</pre>"
            f"{pause_note}"
        )
    else:
        status = compose_status_message(chat_id, include_config=False, include_logs=False)
        logs = session.get('live_log_lines_user', [])
        log_tail = "\n".join(logs[-4:]) if logs else "No logs yet."
        pause_note = " <i>⏸ paused</i>" if is_paused else ""

        # Only show log tail while actively streaming
        if state in ("streaming", "starting", "paused", "reconnecting"):
            return (
                f"{status}"
                f"\n📋 <b>Recent Logs:</b>{pause_note}\n<pre>{esc(log_tail)}</pre>"
            )
        else:
            return status + pause_note


def _get_live_view_keyboard(chat_id: int) -> dict:
    """
    Keyboard for the live-view bubble: Pause/Resume/Refresh + stream controls.
    Mirrors logg.py's create_log_keyboard() pattern.
    """
    session = get_user_session(chat_id)
    state = session.get('streaming_state', 'idle')
    streaming = state in ("streaming", "paused", "starting", "reconnecting")

    with _live_view_lock:
        info = _live_views.get(chat_id, {})
        is_paused_view = not info.get("show", True)

    rows = []

    # Row 1: Live-view controls (always present when registered)
    view_row = []
    if is_paused_view:
        view_row.append({"text": "▶️ Resume Updates", "callback_data": "lv_resume"})
    else:
        view_row.append({"text": "⏸ Pause Updates",  "callback_data": "lv_pause"})
    view_row.append({"text": "🔄 Refresh",  "callback_data": "lv_refresh"})
    view_row.append({"text": "❌ Close",    "callback_data": "lv_close"})
    rows.append(view_row)

    # Row 2: Stream controls (context-aware, same as get_main_keyboard)
    if streaming:
        ctrl = []
        if state == "streaming":
            ctrl.append({"text": "⏸ Pause Stream", "callback_data": "stream_pause"})
        elif state == "paused":
            ctrl.append({"text": "▶️ Resume Stream", "callback_data": "stream_resume"})
        ctrl.append({"text": "⏹ Stop Stream", "callback_data": "stream_abort"})
        rows.append(ctrl)
        if session.get('loop_count', 0) == -1:
            rows.append([{"text": "⏳ Finish After This Loop", "callback_data": "stream_stop_graceful"}])
    else:
        playlist_count = len(session.get('input_url_playlist', []))
        if playlist_count > 0 and session.get('output_url'):
            rows.append([{"text": "🚀 Start Stream", "callback_data": "stream_start"}])

    return {"inline_keyboard": rows}


def register_live_view(chat_id: int, message_id: int, mode: str = "status"):
    """
    Register a message as the live-view bubble for this chat.
    Replaces any existing registration (one live bubble per chat).
    """
    with _live_view_lock:
        _live_views[chat_id] = {
            "message_id": message_id,
            "last_sent": "",
            "show": True,          # True = update edits are enabled
            "paused_at": None,
            "mode": mode,          # "status" or "logs"
        }
    with lock_for(chat_id):
        get_user_session(chat_id)['status_message_id'] = message_id
    logger.info(f"[Chat {chat_id}] Live view registered → msg {message_id} mode={mode}")


def unregister_live_view(chat_id: int):
    with _live_view_lock:
        _live_views.pop(chat_id, None)
    logger.info(f"[Chat {chat_id}] Live view unregistered.")


def set_live_view_show(chat_id: int, show: bool):
    """Toggle live-view updates on/off (Pause/Resume)."""
    with _live_view_lock:
        if chat_id in _live_views:
            _live_views[chat_id]["show"] = show
            _live_views[chat_id]["paused_at"] = None if show else time.strftime('%H:%M:%S')


def _do_live_view_edit(chat_id: int, force: bool = False):
    """
    Build fresh text and enqueue an editMessageText for the live-view bubble.
    Skips if paused (unless force=True) or if text hasn't changed.
    IMPORTANT: Replaces any existing queued edit for this message (cap=1)
    so stale edits never pile up in the outbound queue.
    Also pushes SSE event (for web dashboard clients).
    """
    with _live_view_lock:
        info = _live_views.get(chat_id)
        if not info:
            return
        show = info["show"]
        msg_id = info["message_id"]
        last_sent = info["last_sent"]
        mode = info.get("mode", "status")

    if not show and not force:
        return  # Paused — skip edit (same as logg.py)

    new_text = _build_live_view_text(chat_id, mode)
    kb = _get_live_view_keyboard(chat_id)

    # Only enqueue if content actually changed (avoids "message not modified" errors)
    if new_text == last_sent and not force:
        # Still push SSE even if Telegram text unchanged
        _push_state_sse(chat_id)
        return

    with _live_view_lock:
        if chat_id in _live_views:
            _live_views[chat_id]["last_sent"] = new_text

    edit_payload = edit_message_text(chat_id, msg_id, new_text, reply_markup=kb)

    # Replace (not append) any existing live-view edit in the queue so we never
    # pile up stale edits that Telegram would reject or deliver out of order.
    with _outbound_queue_lock:
        existing = _outbound_message_queue.get(chat_id, [])
        # Remove any previous edit targeting the same message_id
        filtered = [m for m in existing
                    if not (m.get("method") == "editMessageText"
                            and m.get("message_id") == msg_id)]
        filtered.append(edit_payload)
        _outbound_message_queue[chat_id] = filtered

    # Also push SSE snapshot for web clients
    _push_state_sse(chat_id)


def _live_view_updater_loop():
    """
    Background thread — fires every LIVE_VIEW_UPDATE_INTERVAL seconds.
    For each registered live-view, calls _do_live_view_edit().
    On terminal state: does one forced final update then unregisters.
    Also pushes SSE for active sessions without a live-view registered.
    Inspired directly by logg.py's stream_command() update-interval pattern.
    """
    while True:
        time.sleep(LIVE_VIEW_UPDATE_INTERVAL)
        try:
            with _live_view_lock:
                entries = list(_live_views.items())

            for chat_id, info in entries:
                try:
                    session = get_user_session(chat_id)
                    state = session.get('streaming_state', 'idle')

                    _do_live_view_edit(chat_id)

                    # Auto-unregister on terminal state after one forced final update
                    if state in ('idle', 'stopped', 'completed', 'error'):
                        _do_live_view_edit(chat_id, force=True)
                        unregister_live_view(chat_id)
                        # Push final SSE
                        _push_state_sse(chat_id)

                except Exception as e_inner:
                    logger.warning(f"Live view updater error for {chat_id}: {e_inner}")

            # Push SSE for active sessions without a live-view (scheduler streams, etc.)
            try:
                with _live_view_lock:
                    registered = set(_live_views.keys())
                for chat_id, session in list(user_sessions.items()):
                    state = session.get('streaming_state', 'idle')
                    if state in ('streaming', 'starting', 'reconnecting', 'paused') \
                            and chat_id not in registered:
                        _push_state_sse(chat_id)
            except Exception:
                pass

        except Exception as e:
            logger.error(f"Live view updater loop error: {e}")


_live_view_updater_thread = threading.Thread(
    target=_live_view_updater_loop, name="LiveViewUpdater", daemon=True)
_live_view_updater_thread.start()


# Compatibility shim — old code that called register_status_message still works
def register_status_message(chat_id: int, message_id: int):
    register_live_view(chat_id, message_id, mode="status")


def unregister_status_message(chat_id: int):
    unregister_live_view(chat_id)


def pop_pending_status_edit(chat_id: int):
    """
    Legacy shim kept for webhook handler compatibility.
    Now a no-op — edits are enqueued directly by _do_live_view_edit().
    """
    return None


# ──────────────────────────────────────────────
# CONTEXT-AWARE KEYBOARDS
# ──────────────────────────────────────────────
def get_main_keyboard(session: dict):
    """Returns inline keyboard appropriate for current session state."""
    state = session.get('streaming_state', 'idle')
    streaming = state in ("streaming", "paused", "starting", "reconnecting")
    playlist_count = len(session.get('input_url_playlist', []))

    if streaming:
        # Stream control buttons
        btns = []
        row1 = []
        if state == "streaming":
            row1.append({"text": "⏸ Pause", "callback_data": "stream_pause"})
        elif state == "paused":
            row1.append({"text": "▶️ Resume", "callback_data": "stream_resume"})
        row1.append({"text": "⏹ Stop", "callback_data": "stream_abort"})
        btns.append(row1)

        btns.append([
            {"text": "📊 Status", "callback_data": "stream_status"},
            {"text": "📋 Logs", "callback_data": "show_user_logs"},
        ])
        if session.get('loop_count', 0) == -1:
            btns.append([{"text": "⏳ Finish After This Loop", "callback_data": "stream_stop_graceful"}])
        btns.append([{"text": "❔ Help", "callback_data": "show_help"}])
        btns.append([{"text": "⚡ Force Reboot", "callback_data": "force_reboot"}])
    else:
        # Idle mode — full navigation
        btns = []
        # Row 1: Start stream
        if playlist_count > 0 and session.get('output_url'):
            btns.append([{"text": "🚀 Start Stream", "callback_data": "stream_start"}])
        else:
            btns.append([{"text": "⚙️ Quick Setup  (required)", "callback_data": "quick_setup"}])

        # Row 2: Config
        btns.append([
            {"text": f"📜 Playlist ({playlist_count})", "callback_data": "view_playlist"},
            {"text": "⚙️ Settings", "callback_data": "open_settings"},
        ])

        # Row 3: Logo + Schedule
        logo_icon = "🖼✅" if session.get('logo_enabled') else "🖼"
        btns.append([
            {"text": logo_icon + " Logo", "callback_data": "cfg_logo"},
            {"text": "🕒 Schedule", "callback_data": "cfg_schedule"},
        ])

        # Row 4: Utils
        btns.append([
            {"text": "🔄 Reset to Defaults", "callback_data": "confirm_reset"},
            {"text": "❔ Help", "callback_data": "show_help"},
        ])
        # Row 5: Emergency
        btns.append([
            {"text": "⚡ Force Reboot", "callback_data": "force_reboot"},
        ])

    return {"inline_keyboard": btns}


def get_settings_keyboard(session: dict = None):
    """Inline keyboard for settings navigation."""
    ux = (session or {}).get("ux_mode", "send")
    ux_label = "💬 UX: New Message ✅" if ux == "send" else "✏️ UX: Edit In-Place ✅"
    return {"inline_keyboard": [
        [{"text": "📡 Output URL",       "callback_data": "set_output_url"},
         {"text": "🎬 Input Playlist",   "callback_data": "view_playlist"}],
        [{"text": "🎥 Video Codec",      "callback_data": "set_video_codec"},
         {"text": "🔊 Audio Codec",      "callback_data": "set_audio_codec"}],
        [{"text": "📐 Resolution",       "callback_data": "pick_resolution"},
         {"text": "🎞 FPS",              "callback_data": "set_fps"}],
        [{"text": "📶 Video Bitrate",    "callback_data": "set_video_bitrate"},
         {"text": "🔉 Audio Bitrate",    "callback_data": "set_audio_bitrate"}],
        [{"text": "⚡ Preset",            "callback_data": "set_ffmpeg_preset"},
         {"text": "🔁 Loop Count",       "callback_data": "set_loop_count"}],
        [{"text": "🔀 GOP Size",         "callback_data": "set_gop_size"},
         {"text": "📦 Output Format",    "callback_data": "set_output_format"}],
        [{"text": "⏱ Reconnect Delay",   "callback_data": "set_reconnect_delay"},
         {"text": "🔄 Auto-Reconnect",   "callback_data": "toggle_reconnect"}],
        [{"text": "🛑 Stop on Error",    "callback_data": "toggle_stop_on_error"},
         {"text": "⏰ Open Timeout",     "callback_data": "set_open_timeout"}],
        [{"text": "📋 Verbose FFmpeg Log: " + ("✅ On" if (session or {}).get("verbose_ffmpeg_log") else "❌ Off"), "callback_data": "toggle_verbose_log"}],
        [{"text": "🎨 Quality Preset",   "callback_data": "pick_quality_preset"}],
        [{"text": ux_label,              "callback_data": "toggle_ux_mode"}],
        [{"text": "✅ Done",             "callback_data": "settings_done"}],
    ]}


def get_quality_keyboard():
    rows = [[{"text": q.title(), "callback_data": f"apply_quality_{q}"} for q in list(QUALITY_PRESETS.keys())[:3]]]
    rows.append([{"text": q.title(), "callback_data": f"apply_quality_{q}"} for q in list(QUALITY_PRESETS.keys())[3:]])
    rows.append([{"text": "↩ Back", "callback_data": "open_settings"}])
    return {"inline_keyboard": rows}


def get_codec_keyboard(codec_type: str):
    codecs = SUPPORTED_VIDEO_CODECS if codec_type == "video" else SUPPORTED_AUDIO_CODECS
    cb_prefix = "set_vcodec_" if codec_type == "video" else "set_acodec_"
    rows = []
    row = []
    for c in codecs:
        row.append({"text": c, "callback_data": cb_prefix + c})
        if len(row) == 2:
            rows.append(row)
            row = []
    if row:
        rows.append(row)
    rows.append([{"text": "↩ Back to Settings", "callback_data": "open_settings"}])
    return {"inline_keyboard": rows}


def get_preset_keyboard():
    rows = []
    row = []
    for p in FFMPEG_PRESETS:
        row.append({"text": p, "callback_data": "set_preset_" + p})
        if len(row) == 3:
            rows.append(row)
            row = []
    if row:
        rows.append(row)
    rows.append([{"text": "↩ Back", "callback_data": "open_settings"}])
    return {"inline_keyboard": rows}


def get_resolution_keyboard():
    """Inline picker for common resolutions."""
    rows = []
    row = []
    for label, val in RESOLUTION_PRESETS:
        row.append({"text": label, "callback_data": "set_res_" + val})
        if len(row) == 3:
            rows.append(row)
            row = []
    if row:
        rows.append(row)
    rows.append([{"text": "✏️ Custom…", "callback_data": "set_res_custom"}])
    rows.append([{"text": "↩ Back", "callback_data": "open_settings"}])
    return {"inline_keyboard": rows}


def get_logo_pos_keyboard():
    positions = ["top_left", "top_right", "bottom_left", "bottom_right", "center"]
    rows = [[{"text": p.replace("_", " ").title(), "callback_data": f"set_logo_pos_{p}"}] for p in positions]
    rows.append([{"text": "↩ Back", "callback_data": "cfg_logo"}])
    return {"inline_keyboard": rows}


def get_reset_confirm_keyboard():
    return {"inline_keyboard": [
        [{"text": "✅ Yes, Reset Everything", "callback_data": "do_reset"},
         {"text": "❌ Cancel", "callback_data": "show_help"}],
    ]}


# ──────────────────────────────────────────────
# VALIDATION HELPERS
# ──────────────────────────────────────────────
def validate_url(url: str) -> bool:
    try:
        r = urlparse(url)
        return all([r.scheme, r.netloc]) and r.scheme in ['http', 'https', 'rtmp', 'rtsp', 'udp', 'srt', 'file']
    except Exception:
        return False


def validate_resolution(res: str) -> bool:
    if res.lower() == "source":
        return True
    return bool(re.match(r"^\d{3,4}x\d{3,4}$", res))


def parse_bitrate(bitrate_str: str) -> int:
    s = str(bitrate_str).lower()
    try:
        if s.endswith('m'):
            return int(float(s[:-1]) * 1_000_000)
        if s.endswith('k'):
            return int(float(s[:-1]) * 1_000)
        return int(s)
    except ValueError:
        return 1_500_000


def esc(text: str) -> str:
    """HTML-escape a string."""
    return str(text).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


# ──────────────────────────────────────────────
# STATUS & DISPLAY
# ──────────────────────────────────────────────
STATE_EMOJI = {
    "idle": "💤", "starting": "🔄", "streaming": "🔴",
    "paused": "⏸", "stopping": "🛑", "stopped": "⏹",
    "completed": "✅", "error": "❌", "reconnecting": "🔁",
}


def get_uptime(start_time_obj) -> str:
    if not start_time_obj:
        return "—"
    if start_time_obj.tzinfo is None:
        start_time_obj = start_time_obj.replace(tzinfo=datetime.timezone.utc)
    delta = datetime.datetime.now(datetime.timezone.utc) - start_time_obj
    return str(delta).split('.')[0]


def format_settings_display(session: dict) -> str:
    logo_status = "Disabled"
    if session.get('logo_enabled') and session.get('logo_data_bytes'):
        fname = session.get('logo_original_filename', 'Unknown')
        logo_status = f"✅ {esc(fname)}"

    loop = session.get('loop_count', 0)
    loop_disp = "∞ Infinite" if loop == -1 else ("Once" if loop == 0 else f"{loop}×")

    auto_rc = "✅" if session.get('reconnect_on_stream_error') else "❌"
    stop_err = "✅" if session.get('stop_on_error_in_playlist') else "❌"
    verbose_log = "✅ On" if session.get('verbose_ffmpeg_log') else "❌ Off"

    lines = [
        f"📡 <b>Output URL:</b> <code>{esc(session.get('output_url', '—'))}</code>",
        f"📜 <b>Playlist:</b> <code>{len(session.get('input_url_playlist', []))} item(s)</code>",
        "",
        f"🎨 <b>Quality Preset:</b> <code>{esc(session.get('quality_preset', '—'))}</code>",
        f"🎥 <b>Video:</b> <code>{esc(session.get('video_codec'))} | {esc(session.get('resolution'))} | {session.get('fps')}fps | {esc(session.get('video_bitrate'))}</code>",
        f"🔊 <b>Audio:</b> <code>{esc(session.get('audio_codec'))} | {esc(session.get('audio_bitrate'))}</code>",
        f"⚡ <b>Preset:</b> <code>{esc(session.get('ffmpeg_preset'))}</code>  🔀 <b>GOP:</b> <code>{session.get('gop_size')}</code>",
        f"📦 <b>Format:</b> <code>{esc(session.get('output_format', 'flv'))}</code>",
        "",
        f"🔁 <b>Loop:</b> <code>{loop_disp}</code>",
        f"🔄 <b>Auto-Reconnect:</b> {auto_rc}  (delay: <code>{session.get('reconnect_delay_seconds')}s</code>, max: <code>{session.get('max_reconnect_attempts')}</code>)",
        f"🛑 <b>Stop on Error:</b> {stop_err}",
        f"📋 <b>Verbose FFmpeg Log:</b> {verbose_log}",
        f"⏱ <b>Timeouts:</b> open <code>{session.get('open_timeout_seconds')}s</code>  read <code>{session.get('read_timeout_seconds')}s</code>",
        f"🖼 <b>Logo:</b> {logo_status}",
    ]
    if session.get('logo_enabled') and session.get('logo_data_bytes'):
        lines.append(
            f"   Position: <code>{session.get('logo_position')}</code>  "
            f"Scale: <code>{session.get('logo_scale')}</code>  "
            f"Opacity: <code>{session.get('logo_opacity')}</code>"
        )
    return "\n".join(lines)


def compose_status_message(chat_id: int, include_config: bool = False, include_logs: bool = True) -> str:
    session = get_user_session(chat_id)
    state = session.get('streaming_state', 'idle')
    icon = STATE_EMOJI.get(state, "❓")

    lines = [
        f"🤖 <b>Stream Bot v{APP_VERSION}</b>",
        f"{icon} <b>State:</b> <code>{state}</code>",
    ]

    if session.get('error_notification_user'):
        lines.append(f"⚠️ <b>Last Error:</b> <code>{esc(session['error_notification_user'])}</code>")

    if state in ("streaming", "paused", "starting", "stopping", "reconnecting"):
        playlist = session.get('input_url_playlist', [])
        idx = session.get('current_playlist_index', 0)
        cur_url = playlist[idx] if playlist and 0 <= idx < len(playlist) else "N/A"
        loop = session.get('loop_count', 0)
        loop_total = "∞" if loop == -1 else str(max(loop, 1))
        reconnect_att = session.get('reconnect_attempt', 0)

        frames = session.get('frames_encoded', 0)
        bytes_s = session.get('bytes_sent', 0)
        stats_ready = frames > 0 or bytes_s > 0

        stats_lines = []
        if stats_ready:
            stats_lines += [
                f"⏱ <b>Uptime:</b> <code>{get_uptime(session.get('stream_start_time'))}</code>",
                f"🎞 <b>Frames:</b> <code>{frames:,}</code>",
                f"📤 <b>Sent:</b> <code>{bytes_s / (1024*1024):.2f} MB</code>",
            ]

        lines += [
            *stats_lines,
            f"🎬 <b>Input:</b> <code>{esc(cur_url)}</code> ({idx+1}/{len(playlist)})",
            f"🔁 <b>Loop:</b> <code>{session.get('current_loop_iteration', 0)+1}/{loop_total}</code>",
        ]
        if reconnect_att > 0:
            lines.append(f"🔄 <b>Reconnect attempts:</b> <code>{reconnect_att}</code>")

    if include_config or state in ("idle", "stopped", "completed", "error"):
        lines += ["", "⚙️ <b>Configuration:</b>", format_settings_display(session)]

    if include_logs and state in ("streaming", "paused", "starting", "reconnecting"):
        user_logs = session.get('live_log_lines_user', [])
        last_logs = "\n".join(user_logs[-4:]) if user_logs else "No logs yet."
        lines += [f"\n📋 <b>Recent Logs:</b>\n<pre>{esc(last_logs)}</pre>"]

    return "\n".join(lines)


# ──────────────────────────────────────────────
# STREAM STATE CHANGE NOTIFICATION
# ──────────────────────────────────────────────
def notify_state_change(chat_id: int, new_state: str, extra_msg: str = ""):
    """Push a message to user when stream state changes."""
    session = get_user_session(chat_id)
    icon = STATE_EMOJI.get(new_state, "❓")
    text = f"{icon} <b>Stream state changed:</b> <code>{new_state}</code>"
    if extra_msg:
        text += f"\n{extra_msg}"
    text += "\n\n" + compose_status_message(chat_id)
    kb = get_main_keyboard(session)
    # Webhook-only: queue outbound message to be returned on next webhook response
    push_message(chat_id, text, reply_markup=kb)
    enqueue_outbound_message(chat_id, send_message(chat_id, text, reply_markup=kb))
    # Push SSE state event with full status snapshot
    _push_state_sse(chat_id)


def _push_state_sse(chat_id: int):
    """Build and push a full real-time state snapshot to SSE subscribers."""
    try:
        session = get_user_session(chat_id)
        state = session.get('streaming_state', 'idle')
        uptime = 0
        if session.get('stream_start_time'):
            try:
                uptime = int((datetime.datetime.now(datetime.timezone.utc) - session['stream_start_time']).total_seconds())
            except Exception:
                pass
        # Build keyboard state for client-side button rendering
        kb = get_main_keyboard(session)
        push_sse_event(chat_id, {
            "type": "state",
            "state": state,
            "icon": STATE_EMOJI.get(state, "❓"),
            "frames_encoded": session.get('frames_encoded', 0),
            "bytes_sent": session.get('bytes_sent', 0),
            "uptime_seconds": uptime,
            "uptime_str": get_uptime(session.get('stream_start_time')),
            "reconnect_attempt": session.get('reconnect_attempt', 0),
            "error": session.get('error_notification_user', ''),
            "active_output_url": session.get('active_output_url', ''),
            "playlist_index": session.get('current_playlist_index', 0),
            "playlist_count": len(session.get('input_url_playlist', [])),
            "loop_iteration": session.get('current_loop_iteration', 0),
            "status_text": compose_status_message(chat_id, include_config=False),
            "keyboard": kb,
            "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        })
    except Exception as e:
        logger.warning(f"_push_state_sse error for {chat_id}: {e}")


def update_streaming_state(chat_id: int, new_state: str, lock, session: dict, extra_msg: str = ""):
    """Thread-safe state update + push notification if state changed."""
    old_state = session.get('streaming_state')
    session['streaming_state'] = new_state
    should_notify = (old_state != new_state)

    if should_notify:
        # Notify outside the lock
        threading.Thread(
            target=notify_state_change,
            args=(chat_id, new_state, extra_msg),
            daemon=True
        ).start()
    else:
        # Still push SSE stats even if state didn't change
        threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start()


# ──────────────────────────────────────────────
# WATCHDOG THREAD
# ──────────────────────────────────────────────
def watchdog_thread_target(chat_id: int):
    """Monitors the stream thread for freezes and notifies user."""
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    STALL_THRESHOLD = 30  # seconds without a new frame = stall

    logger.info(f"[Chat {chat_id}] Watchdog started.")
    while True:
        time.sleep(5)
        with lock:
            state = session.get('streaming_state')
            # Exit watchdog if stream is in any terminal or stopping state
            if state not in ("streaming", "paused", "reconnecting", "starting"):
                break

            last_frame = session.get('last_frame_time')
            thread_ref = session.get('stream_thread_ref')

        # Check if stream thread died unexpectedly
        if thread_ref and not thread_ref.is_alive():
            with lock:
                cur_state = session.get('streaming_state')
            # Only mark error if not already in a clean terminal/stopping state
            if cur_state in ("streaming", "paused", "reconnecting", "starting"):
                logger.error(f"[Chat {chat_id}] Watchdog: stream thread died unexpectedly!")
                append_user_live_log(chat_id, "⚠️ Stream thread died unexpectedly. Marking as error.")
                with lock:
                    session['streaming_state'] = "error"
                    session['error_notification_user'] = "Stream thread terminated unexpectedly."
                notify_state_change(chat_id, "error", "The stream thread crashed. Use /stream to restart.")
            break

        # Check for frame stall (only while actively streaming)
        if state == "streaming" and last_frame:
            stall_seconds = (datetime.datetime.now() - last_frame).total_seconds()
            if stall_seconds > STALL_THRESHOLD:
                append_user_live_log(chat_id, f"⚠️ Watchdog: No frames for {stall_seconds:.0f}s — possible freeze!")
                enqueue_outbound_message(chat_id, send_message(
                    chat_id,
                    f"⚠️ <b>Stream may be frozen!</b>\nNo frames for <code>{stall_seconds:.0f}s</code>.\n"
                    f"Use /abort to stop or wait for auto-reconnect.",
                    reply_markup=get_main_keyboard(session)
                ))

    logger.info(f"[Chat {chat_id}] Watchdog exiting.")


# ──────────────────────────────────────────────
# CORE STREAM ENGINE
# ──────────────────────────────────────────────
def stream_engine_thread_target(chat_id: int):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]

    def _cleanup_pyav():
        append_user_live_log(chat_id, "Cleaning up PyAV resources…")
        with lock:
            objs = session.get('pyav_objects', {})
            for key in ('output_container', 'input_container'):
                container = objs.get(key)
                if container:
                    try:
                        container.close()
                    except Exception as e:
                        append_user_live_log(chat_id, f"Error closing {key}: {e}")
                objs[key] = None
            objs['video_out_stream'] = None
            objs['audio_out_stream'] = None
            objs['logo_image_pil'] = None

    active_output_container = None

    # ── Raw FFmpeg/libav log capture via queue ────────────────────────────────
    # PyAV's av.logging.set_log_callback installs a C-level callback. We route
    # every libav message into a thread-safe queue and drain it in a background
    # thread so we never call Python from a C signal context.
    _fflog_queue: queue.Queue = queue.Queue(maxsize=500)
    _fflog_stop = threading.Event()

    def _fflog_drain():
        _LEVELS = {0:"PANIC",8:"FATAL",16:"ERROR",24:"WARN",32:"INFO",40:"VERBOSE"}
        _NOISE = ("Past duration", "deprecated pixel format", "DTS", "PTS", "non monotonous")
        _last = ""
        while not _fflog_stop.is_set() or not _fflog_queue.empty():
            try:
                level, msg = _fflog_queue.get(timeout=0.3)
            except queue.Empty:
                continue
            msg = msg.strip()
            if not msg or msg == _last:
                continue
            # Verbose mode off → only pass warnings/errors (level <= 24)
            verbose_on = session.get('verbose_ffmpeg_log', False)
            if not verbose_on and level > 24:
                continue
            # Skip high-volume noise regardless of verbose setting
            if any(s in msg for s in _NOISE):
                continue
            _last = msg
            lvl = _LEVELS.get(level, f"L{level}")
            prefix = f"[ffmpeg/{lvl}]" if level <= 24 else "[ffmpeg]"
            append_user_live_log(chat_id, f"{prefix} {msg}")

    def _fflog_callback(level, message):
        try:
            _fflog_queue.put_nowait((level, message))
        except queue.Full:
            pass

    _fflog_thread = threading.Thread(target=_fflog_drain, name=f"FFlogDrain-{chat_id}", daemon=True)
    _fflog_thread.start()
    try:
        av.logging.set_log_level(av.logging.VERBOSE)
        av.logging.set_log_callback(_fflog_callback)
    except Exception as _e_avlog:
        append_user_live_log(chat_id, f"[warn] av log callback unavailable: {_e_avlog}")
    # ─────────────────────────────────────────────────────────────────────────

    try:
        with lock:
            session['streaming_state'] = "starting"
            session['error_notification_user'] = ""
            session['stream_start_time'] = datetime.datetime.now(datetime.timezone.utc)
            session['frames_encoded'] = 0
            session['bytes_sent'] = 0
            session['current_loop_iteration'] = 0
            session['current_playlist_index'] = 0
            session['stop_gracefully_flag'] = False
            session['reconnect_attempt'] = 0
            session['last_frame_time'] = None
            session['live_log_lines_user'] = []   # clear stale logs from previous run
            session['pyav_objects'] = {
                "input_container": None, "output_container": None,
                "video_out_stream": None, "audio_out_stream": None,
                "logo_image_pil": None,
            }
        # Also reset last_sent so the live bubble always refreshes immediately
        with _live_view_lock:
            if chat_id in _live_views:
                _live_views[chat_id]["last_sent"] = ""

        append_user_live_log(chat_id, f"Stream engine started → {session['output_url']}")

        # Notify state change
        notify_state_change(chat_id, "starting")

        # --- Open output container ---
        # Always read fresh from session so any /set output_url change takes effect.
        with lock:
            output_url = session['output_url']
            output_format = session.get('output_format', 'flv')
            open_timeout = session.get('open_timeout_seconds', 15)
        # Also store the active_output_url in session for status display
        with lock:
            session['active_output_url'] = output_url
        append_user_live_log(chat_id, f"Opening output: {output_url} [{output_format}]")
        try:
            active_output_container = av.open(
                output_url, mode='w', format=output_format,
                timeout=open_timeout
            )
            with lock:
                session['pyav_objects']['output_container'] = active_output_container
            append_user_live_log(chat_id, f"Output container opened ({output_format}).")
        except Exception as e:
            raise Exception(f"Failed to open output stream: {e}")

        # --- Load logo ---
        logo_pil_original = None
        if session.get('logo_enabled') and session.get('logo_data_bytes'):
            try:
                logo_pil_original = Image.open(io.BytesIO(session['logo_data_bytes'])).convert("RGBA")
                opacity = session.get('logo_opacity', 1.0)
                if opacity < 1.0:
                    alpha = logo_pil_original.split()[-1]
                    alpha = ImageEnhance.Brightness(alpha).enhance(opacity)
                    logo_pil_original.putalpha(alpha)
                with lock:
                    session['pyav_objects']['logo_image_pil'] = logo_pil_original
                append_user_live_log(chat_id, f"Logo loaded: {session.get('logo_original_filename', 'N/A')}")
            except Exception as e:
                append_user_live_log(chat_id, f"Logo load error: {e}. Skipping logo.")
                logo_pil_original = None

        # --- Main loop ---
        total_loops = session.get('loop_count', 0)
        playlist = session.get('input_url_playlist', [])

        if not playlist:
            raise Exception("Playlist is empty. Add URLs with /playlist add <url>")

        active_video_out_stream = None
        active_audio_out_stream = None
        output_configured = False
        current_logo_resized = None
        logo_pos_x, logo_pos_y = 0, 0

        while True:
            with lock:
                if session.get('stop_gracefully_flag') or session['streaming_state'] == "stopping":
                    break
                current_loop_iter = session['current_loop_iteration']
                # Re-read total_loops and playlist each iteration so live changes are picked up
                total_loops = session.get('loop_count', 0)
                playlist = list(session.get('input_url_playlist', []))

            if total_loops != -1 and current_loop_iter >= max(total_loops, 1):
                append_user_live_log(chat_id, f"Completed all {max(total_loops, 1)} loop(s).")
                break

            append_user_live_log(chat_id, f"Loop iteration {current_loop_iter + 1}")
            with lock:
                session['current_playlist_index'] = 0

            while True:
                with lock:
                    cur_idx = session['current_playlist_index']
                    should_stop = session.get('stop_gracefully_flag') or session['streaming_state'] == "stopping"
                    playlist_done = cur_idx >= len(playlist)
                    cur_playlist = list(session['input_url_playlist'])

                if should_stop or playlist_done:
                    break

                current_input_url = cur_playlist[cur_idx]
                append_user_live_log(chat_id, f"Input [{cur_idx+1}/{len(cur_playlist)}]: {current_input_url}")

                _input_container = None
                reconnect_on_error = session.get('reconnect_on_stream_error', True)
                reconnect_delay = session.get('reconnect_delay_seconds', 5)
                max_reconnects = session.get('max_reconnect_attempts', 3)
                read_timeout = session.get('read_timeout_seconds', 30)

                try:
                    _input_container = av.open(current_input_url, timeout=open_timeout)
                    with lock:
                        session['pyav_objects']['input_container'] = _input_container
                        session['reconnect_attempt'] = 0

                    in_v_streams = _input_container.streams.video
                    in_a_streams = _input_container.streams.audio
                    if not in_v_streams:
                        raise Exception("No video stream found in input.")

                    in_v_s = in_v_streams[0]
                    target_w, target_h = in_v_s.width, in_v_s.height
                    if session.get('resolution', 'source') != 'source':
                        try:
                            target_w, target_h = map(int, session['resolution'].split('x'))
                        except Exception:
                            pass

                    # Configure output streams (once)
                    if not output_configured:
                        fps = session.get('fps', 30)
                        active_video_out_stream = active_output_container.add_stream(
                            session.get('video_codec', 'libx264'), rate=fps
                        )
                        active_video_out_stream.width = target_w
                        active_video_out_stream.height = target_h
                        active_video_out_stream.pix_fmt = session.get('video_pix_fmt', 'yuv420p')
                        active_video_out_stream.codec_context.gop_size = session.get('gop_size', fps * 2)
                        active_video_out_stream.codec_context.bit_rate = parse_bitrate(session.get('video_bitrate', '1500k'))
                        tc = session.get('video_thread_count', 0)
                        if tc > 0:
                            active_video_out_stream.codec_context.thread_count = tc
                        active_video_out_stream.codec_context.thread_type = "AUTO"
                        preset = session.get('ffmpeg_preset', 'medium')
                        if session.get('video_codec', 'libx264') not in ('copy',):
                            active_video_out_stream.codec_context.options = {'preset': preset}

                        with lock:
                            session['pyav_objects']['video_out_stream'] = active_video_out_stream

                        if in_a_streams:
                            in_a_s = in_a_streams[0]
                            a_rate = session.get('audio_sample_rate') or in_a_s.rate
                            active_audio_out_stream = active_output_container.add_stream(
                                session.get('audio_codec', 'aac'), rate=a_rate
                            )
                            active_audio_out_stream.codec_context.bit_rate = parse_bitrate(session.get('audio_bitrate', '128k'))
                            with lock:
                                session['pyav_objects']['audio_out_stream'] = active_audio_out_stream

                        # Logo scaling
                        if logo_pil_original:
                            logo_scale = session.get('logo_scale', 0.10)
                            margin = session.get('logo_margin_px', 10)
                            logo_h = int(target_h * logo_scale)
                            logo_w = int(logo_h * (logo_pil_original.width / logo_pil_original.height))
                            logo_w = min(logo_w, target_w)
                            logo_h = min(logo_h, target_h)
                            if logo_w > 0 and logo_h > 0:
                                current_logo_resized = logo_pil_original.resize((logo_w, logo_h), Image.Resampling.LANCZOS)
                                pos = session.get('logo_position', 'top_right')
                                if pos == 'top_right':
                                    logo_pos_x, logo_pos_y = target_w - logo_w - margin, margin
                                elif pos == 'top_left':
                                    logo_pos_x, logo_pos_y = margin, margin
                                elif pos == 'bottom_left':
                                    logo_pos_x, logo_pos_y = margin, target_h - logo_h - margin
                                elif pos == 'bottom_right':
                                    logo_pos_x, logo_pos_y = target_w - logo_w - margin, target_h - logo_h - margin
                                elif pos == 'center':
                                    logo_pos_x, logo_pos_y = (target_w - logo_w) // 2, (target_h - logo_h) // 2

                        output_configured = True
                        append_user_live_log(chat_id, f"Output streams configured: {target_w}x{target_h}@{fps}fps")

                    with lock:
                        if session['streaming_state'] != "paused":
                            old_st = session['streaming_state']
                            session['streaming_state'] = "streaming"
                            if old_st != "streaming":
                                threading.Thread(target=notify_state_change, args=(chat_id, "streaming"), daemon=True).start()

                    # --- Packet loop ---
                    last_read_time = time.time()
                    for packet in _input_container.demux(video=0, audio=0 if in_a_streams else -1):
                        with lock:
                            if session['streaming_state'] == "stopping":
                                break
                            while session['streaming_state'] == "paused":
                                lock.release()
                                time.sleep(0.2)
                                lock.acquire()
                            if session['streaming_state'] == "stopping":
                                break

                        # Read timeout check
                        if time.time() - last_read_time > read_timeout:
                            raise Exception(f"Read timeout after {read_timeout}s of no packets.")
                        last_read_time = time.time()

                        if packet.dts is None:
                            continue

                        if packet.stream.type == 'video' and packet.stream.index == in_v_s.index:
                            for frame in packet.decode():
                                if not frame.width or not frame.height:
                                    continue
                                if frame.width != target_w or frame.height != target_h:
                                    frame = frame.reformat(target_w, target_h, format='yuv420p')

                                if current_logo_resized:
                                    try:
                                        pil_f = frame.to_image()
                                        ax = max(0, min(logo_pos_x, pil_f.width - current_logo_resized.width))
                                        ay = max(0, min(logo_pos_y, pil_f.height - current_logo_resized.height))
                                        pil_f.paste(current_logo_resized, (ax, ay), current_logo_resized)
                                        new_pts = frame.pts
                                        frame = av.VideoFrame.from_image(pil_f)
                                        frame.pts = new_pts
                                    except Exception as e_logo:
                                        pass

                                # Count decoded input frames immediately.
                                # The encoder buffers many frames before emitting
                                # the first output packet, so counting only encoded
                                # output kept stats stuck at 0 for a long time.
                                with lock:
                                    session['frames_encoded'] += 1
                                    session['last_frame_time'] = datetime.datetime.now()
                                _fc = session['frames_encoded']
                                # Push live stats every 30 decoded frames (~1s at 30fps)
                                if _fc % 30 == 0:
                                    threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start()

                                for out_pkt in active_video_out_stream.encode(frame):
                                    active_output_container.mux(out_pkt)
                                    with lock:
                                        session['bytes_sent'] += out_pkt.size

                        elif active_audio_out_stream and packet.stream.type == 'audio' and in_a_streams and packet.stream.index == in_a_streams[0].index:
                            for frame in packet.decode():
                                if frame.pts is None:
                                    continue
                                for out_pkt in active_audio_out_stream.encode(frame):
                                    active_output_container.mux(out_pkt)
                                    with lock:
                                        session['bytes_sent'] += out_pkt.size

                    append_user_live_log(chat_id, f"Finished input: {current_input_url}")

                except Exception as e_input:
                    tb = traceback.format_exc()
                    err_msg = str(e_input)
                    # Strip any log-format prefix that might have leaked in
                    if "] " in err_msg and "[" in err_msg[:40]:
                        err_msg = err_msg.split("] ", 1)[-1] if "] " in err_msg else err_msg
                    append_user_live_log(chat_id, f"Error on input {current_input_url}: {err_msg}")
                    with lock:
                        session['error_notification_user'] = err_msg[:300]

                    with lock:
                        should_stop_now = session['streaming_state'] == "stopping"
                        reconnect_on = session.get('reconnect_on_stream_error', True)
                        cur_attempt = session.get('reconnect_attempt', 0)
                        max_att = session.get('max_reconnect_attempts', 3)
                        stop_on_err = session.get('stop_on_error_in_playlist', True)

                    if should_stop_now:
                        break

                    if reconnect_on and cur_attempt < max_att:
                        with lock:
                            session['reconnect_attempt'] = cur_attempt + 1
                            old_st = session['streaming_state']
                            if session['streaming_state'] not in ('stopping',):
                                session['streaming_state'] = "reconnecting"
                        append_user_live_log(chat_id, f"Reconnecting in {reconnect_delay}s (attempt {cur_attempt+1}/{max_att})…")
                        if old_st != "reconnecting":
                            notify_state_change(chat_id, "reconnecting", f"Attempt {cur_attempt+1}/{max_att}. Error: {esc(err_msg)}")
                        # Close bad input
                        if _input_container:
                            try: _input_container.close()
                            except Exception: pass
                        # Sleep in small increments so abort flag is checked promptly
                        for _ in range(int(reconnect_delay * 5)):
                            with lock:
                                if session['streaming_state'] == 'stopping':
                                    break
                            time.sleep(0.2)
                        with lock:
                            if session['streaming_state'] == 'stopping':
                                break
                        # Retry same URL (re-enter inner loop iteration)
                        continue
                    elif stop_on_err:
                        append_user_live_log(chat_id, "Stopping due to error (stop_on_error=True).")
                        with lock:
                            session['streaming_state'] = "stopping"
                        break
                    else:
                        append_user_live_log(chat_id, "Skipping to next playlist item.")

                finally:
                    if _input_container:
                        try:
                            _input_container.close()
                        except Exception:
                            pass
                    with lock:
                        if session['pyav_objects'].get('input_container') is _input_container:
                            session['pyav_objects']['input_container'] = None

                with lock:
                    if session['streaming_state'] == "stopping":
                        break
                    session['current_playlist_index'] += 1

            with lock:
                if session['streaming_state'] == "stopping":
                    break
                session['current_loop_iteration'] += 1

        append_user_live_log(chat_id, "All items/loops processed or stopped.")

    except Exception as e_fatal:
        tb = traceback.format_exc()
        append_user_live_log(chat_id, f"Fatal error: {e_fatal}")
        logger.error(f"[Chat {chat_id}] Fatal stream error: {e_fatal}\n{tb}")
        with lock:
            session['error_notification_user'] = str(e_fatal)
            session['streaming_state'] = "error"
        notify_state_change(chat_id, "error", f"Fatal error: {esc(str(e_fatal))}")

    finally:
        # Shut down raw ffmpeg log capture
        try:
            av.logging.restore_default_callback()
        except Exception:
            pass
        _fflog_stop.set()

        append_user_live_log(chat_id, "Finalizing stream…")

        # Flush encoders
        _final_output = None
        _final_vid = None
        _final_aud = None
        with lock:
            objs = session.get('pyav_objects', {})
            _final_output = objs.get('output_container')
            _final_vid = objs.get('video_out_stream')
            _final_aud = objs.get('audio_out_stream')

        if _final_output:
            append_user_live_log(chat_id, "Flushing encoders…")
            for stream in [_final_vid, _final_aud]:
                if stream:
                    try:
                        for pkt in stream.encode():
                            try:
                                _final_output.mux(pkt)
                            except Exception:
                                pass
                    except Exception as fe:
                        append_user_live_log(chat_id, f"Flush error: {fe}")

        _cleanup_pyav()

        with lock:
            cur = session['streaming_state']
            if cur == "stopping":
                final_state = "stopped"
                session['error_notification_user'] = ""  # clear error on clean stop
            elif cur == "error":
                final_state = "error"
            elif session.get('error_notification_user'):
                final_state = "error"
            else:
                final_state = "completed"
                session['error_notification_user'] = ""
            session['streaming_state'] = final_state
            session['stream_thread_ref'] = None
            session['watchdog_thread_ref'] = None
            session['stop_gracefully_flag'] = False

        append_user_live_log(chat_id, f"Stream ended. Final state: {final_state}.")
        notify_state_change(chat_id, final_state,
            "Stream completed successfully! ✅" if final_state == "completed" else
            "Stream was stopped." if final_state == "stopped" else
            f"Error: {esc(session.get('error_notification_user', ''))}")


# ──────────────────────────────────────────────
# STREAM CONTROL HANDLERS
# ──────────────────────────────────────────────
async def start_stream_handler(chat_id: int, message_id_to_edit: int = None):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]

    def _reply(text, kb):
        if message_id_to_edit:
            return edit_message_text(chat_id, message_id_to_edit, text, reply_markup=kb)
        return send_message(chat_id, text, reply_markup=kb)

    with lock:
        state = session['streaming_state']
        if state in ("streaming", "paused", "starting", "reconnecting"):
            return _reply(
                f"⚠️ Stream already active (state: <code>{state}</code>).\n"
                f"Use /abort to stop it first.",
                get_main_keyboard(session))
        if not session.get('input_url_playlist'):
            return _reply(
                "📜 <b>Playlist is empty.</b>\nAdd at least one URL:\n"
                "<code>/playlist add &lt;url&gt;</code>",
                get_main_keyboard(session))
        if not session.get('output_url') or session['output_url'] == DEFAULT_USER_SETTINGS['output_url']:
            return _reply(
                "📡 <b>Output URL not configured.</b>\n"
                "Set it with: <code>/set output_url rtmp://your-url/key</code>",
                get_main_keyboard(session))
        session['streaming_state'] = "starting"
        session['error_notification_user'] = ""

    t_stream = threading.Thread(target=stream_engine_thread_target, args=(chat_id,),
                                name=f"Stream-{chat_id}", daemon=True)
    t_watch = threading.Thread(target=watchdog_thread_target, args=(chat_id,),
                               name=f"Watchdog-{chat_id}", daemon=True)
    with lock:
        session['stream_thread_ref'] = t_stream
        session['watchdog_thread_ref'] = t_watch

    t_stream.start()
    await asyncio.sleep(0.3)
    t_watch.start()

    return _reply(
        "🔄 <b>Stream starting…</b>\n\n"
        "I'll notify you when it's live.\n\n"
        "<b>Commands while streaming:</b>\n"
        "  /pause — pause the stream\n"
        "  /resume — resume\n"
        "  /abort — stop the stream\n"
        "  /status — current status",
        get_main_keyboard(session))


async def pause_stream_handler(chat_id: int, message_id: int = None):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    with lock:
        if session['streaming_state'] == "streaming":
            session['streaming_state'] = "paused"
            ok = True
        else:
            ok = False
            st = session['streaming_state']

    def _reply(text, kb):
        if message_id:
            return edit_message_text(chat_id, message_id, text, reply_markup=kb)
        return send_message(chat_id, text, reply_markup=kb)

    if ok:
        append_user_live_log(chat_id, "Paused by user.")
        threading.Thread(target=notify_state_change, args=(chat_id, "paused"), daemon=True).start()
        return _reply("⏸ <b>Stream paused.</b>\nUse /resume to continue or /abort to stop.",
                      get_main_keyboard(session))
    else:
        return _reply(f"ℹ️ Cannot pause — state is <code>{st}</code>.", get_main_keyboard(session))


async def resume_stream_handler(chat_id: int, message_id: int = None):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    with lock:
        if session['streaming_state'] == "paused":
            session['streaming_state'] = "streaming"
            ok = True
        else:
            ok = False
            st = session['streaming_state']

    def _reply(text, kb):
        if message_id:
            return edit_message_text(chat_id, message_id, text, reply_markup=kb)
        return send_message(chat_id, text, reply_markup=kb)

    if ok:
        append_user_live_log(chat_id, "Resumed by user.")
        threading.Thread(target=notify_state_change, args=(chat_id, "streaming"), daemon=True).start()
        return _reply("▶️ <b>Stream resumed.</b>", get_main_keyboard(session))
    else:
        return _reply(f"ℹ️ Cannot resume — state is <code>{st}</code>.", get_main_keyboard(session))


async def force_reboot_handler(chat_id: int, message_id: int = None):
    """
    Force Reboot — last-resort recovery tool.
    Kills any running stream thread, wipes ALL runtime state back to defaults
    while preserving user settings (URLs, codec, etc.), clears all queues,
    unregisters the live view, and returns the bot to a clean idle state.
    """
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]

    append_user_live_log(chat_id, "⚡ Force Reboot initiated by user.")

    # 1. Signal any running thread to stop
    with lock:
        session['streaming_state'] = "stopping"
        session['stop_gracefully_flag'] = True
        thread_ref = session.get('stream_thread_ref')

    # 2. Wait briefly for thread to die
    if thread_ref and thread_ref.is_alive():
        thread_ref.join(timeout=5.0)

    # 3. Hard-close any open PyAV containers
    try:
        with lock:
            objs = session.get('pyav_objects', {})
            for key in ('output_container', 'input_container'):
                c = objs.get(key)
                if c:
                    try: c.close()
                    except Exception: pass
                objs[key] = None
            objs['video_out_stream'] = None
            objs['audio_out_stream'] = None
            objs['logo_image_pil'] = None
    except Exception as e:
        logger.warning(f"[Reboot] PyAV close error: {e}")

    # 4. Reset ALL runtime state — keep user settings intact
    with lock:
        for k, v in DEFAULT_SESSION_RUNTIME_STATE.items():
            session[k] = dict(v) if isinstance(v, dict) else (list(v) if isinstance(v, list) else v)

    # 5. Clear queues and live view
    unregister_live_view(chat_id)
    with _outbound_queue_lock:
        _outbound_message_queue.pop(chat_id, None)

    _push_state_sse(chat_id)
    append_user_live_log(chat_id, "Force Reboot complete — bot is in clean idle state.")

    reply_text = (
        "⚡ <b>Force Reboot Complete!</b>\n\n"
        "All stream threads killed, state wiped to clean idle.\n"
        "Your settings (URLs, codec, etc.) are preserved.\n\n"
        + compose_status_message(chat_id, include_config=False)
    )

    def _reply(text, kb):
        if message_id:
            return edit_message_text(chat_id, message_id, text, reply_markup=kb)
        return send_message(chat_id, text, reply_markup=kb)

    return _reply(reply_text, get_main_keyboard(session))


async def abort_stream_handler(chat_id: int, message_id: int = None):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    thread_ref = None

    with lock:
        state = session['streaming_state']
        if state in ("streaming", "paused", "starting", "reconnecting", "stopping"):
            session['streaming_state'] = "stopping"
            session['stop_gracefully_flag'] = True
            thread_ref = session.get('stream_thread_ref')
            aborted = True
        else:
            aborted = False
            st = state

    def _reply(text, kb):
        if message_id:
            return edit_message_text(chat_id, message_id, text, reply_markup=kb)
        return send_message(chat_id, text, reply_markup=kb)

    if not aborted:
        return _reply(f"ℹ️ No active stream (state: <code>{st}</code>).", get_main_keyboard(session))

    append_user_live_log(chat_id, "Abort requested by user.")
    if thread_ref and thread_ref.is_alive():
        thread_ref.join(timeout=10.0)
        if thread_ref.is_alive():
            append_user_live_log(chat_id, "Warning: thread did not stop cleanly in 10s — forcing.")

    with lock:
        # Force stopped state regardless of what thread set it to
        if session['streaming_state'] in ("stopping", "reconnecting", "starting", "streaming", "paused"):
            session['streaming_state'] = "stopped"
        session['stream_thread_ref'] = None
        session['watchdog_thread_ref'] = None
        session['stop_gracefully_flag'] = False
        session['error_notification_user'] = ""   # clear error on clean stop

    _push_state_sse(chat_id)
    return _reply("⏹ <b>Stream stopped.</b>\n\n" + compose_status_message(chat_id),
                  get_main_keyboard(session))


# ──────────────────────────────────────────────
# PLAYLIST HANDLER
# ──────────────────────────────────────────────
async def handle_playlist_command(chat_id: int, text: str):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    parts = text.split(maxsplit=2)
    action = parts[1].lower() if len(parts) > 1 else "show"

    with lock:
        playlist = session.setdefault('input_url_playlist', [])

        if action == "add":
            if len(parts) < 3:
                msg = "⚠️ Usage: <code>/playlist add &lt;url&gt;</code>"
            else:
                url = parts[2].strip()
                if validate_url(url):
                    playlist.append(url)
                    msg = f"✅ Added to playlist.\n<code>{esc(url)}</code>\nTotal: <b>{len(playlist)}</b>"
                else:
                    msg = f"❌ Invalid URL.\nMust start with http/https/rtmp/rtsp/udp/srt/file.\n<code>{esc(url)}</code>"

        elif action == "remove":
            if len(parts) < 3:
                msg = "⚠️ Usage: <code>/playlist remove &lt;index|last&gt;</code>"
            else:
                idx_str = parts[2].strip()
                removed = None
                if idx_str.lower() == "last" and playlist:
                    removed = playlist.pop()
                elif idx_str.isdigit():
                    i = int(idx_str) - 1
                    if 0 <= i < len(playlist):
                        removed = playlist.pop(i)
                if removed:
                    msg = f"🗑 Removed: <code>{esc(removed[:80])}</code>\nRemaining: {len(playlist)}"
                else:
                    msg = "❌ Invalid index or empty playlist."

        elif action == "clear":
            count = len(playlist)
            session['input_url_playlist'] = []
            msg = f"🗑 Playlist cleared — {count} item(s) removed."

        elif action == "show":
            if playlist:
                items = "\n".join(f"  <code>{i+1}.</code> {esc(url)}" for i, url in enumerate(playlist))
                msg = f"📜 <b>Playlist ({len(playlist)} items):</b>\n{items}"
            else:
                msg = ("📜 <b>Playlist is empty.</b>\n"
                       "Add a URL:\n<code>/playlist add &lt;url&gt;</code>")

        else:
            msg = ("📜 <b>Playlist Commands:</b>\n"
                   "  <code>/playlist add &lt;url&gt;</code>\n"
                   "  <code>/playlist remove &lt;index|last&gt;</code>\n"
                   "  <code>/playlist clear</code>\n"
                   "  <code>/playlist show</code>")

    return send_message(chat_id, msg, reply_markup=get_main_keyboard(session))


# ──────────────────────────────────────────────
# GENERIC SET COMMAND  /set <field> <value>
# ──────────────────────────────────────────────
SETTABLE_FIELDS = {
    "output_url":               (validate_url,                         "RTMP/HTTP output URL"),
    "video_codec":              (lambda x: x in SUPPORTED_VIDEO_CODECS, f"Video codec. Options: {', '.join(SUPPORTED_VIDEO_CODECS)}"),
    "audio_codec":              (lambda x: x in SUPPORTED_AUDIO_CODECS, f"Audio codec. Options: {', '.join(SUPPORTED_AUDIO_CODECS)}"),
    "resolution":               (validate_resolution,                   "e.g. 1280x720 or source"),
    "fps":                      (lambda x: x.isdigit() and 1 <= int(x) <= 120, "FPS (1–120)"),
    "video_bitrate":            (lambda x: True,                        "e.g. 1500k or 3M"),
    "audio_bitrate":            (lambda x: True,                        "e.g. 128k"),
    "ffmpeg_preset":            (lambda x: x in FFMPEG_PRESETS,         f"Options: {', '.join(FFMPEG_PRESETS)}"),
    "gop_size":                 (lambda x: x.isdigit(),                 "Integer, e.g. 60"),
    "loop_count":               (lambda x: x.lstrip('-').isdigit(),     "0=once, -1=infinite, N=N times"),
    "output_format":            (lambda x: x in OUTPUT_FORMATS,         f"Options: {', '.join(OUTPUT_FORMATS)}"),
    "reconnect_delay_seconds":  (lambda x: x.isdigit(),                 "Seconds between reconnect attempts"),
    "max_reconnect_attempts":   (lambda x: x.isdigit(),                 "Max number of reconnect attempts"),
    "open_timeout_seconds":     (lambda x: x.isdigit(),                 "Connection open timeout in seconds"),
    "read_timeout_seconds":     (lambda x: x.isdigit(),                 "Packet read timeout in seconds"),
    "video_pix_fmt":            (lambda x: True,                        "Pixel format, e.g. yuv420p"),
    "video_thread_count":       (lambda x: x.isdigit(),                 "Thread count (0=auto)"),
    "audio_sample_rate":        (lambda x: x.isdigit(),                 "Sample rate (0=source), e.g. 44100"),
    "audio_channels":           (lambda x: x.isdigit(),                 "Channels (0=source), e.g. 2"),
    "logo_position":            (lambda x: x in LOGO_POSITIONS,         f"Options: {', '.join(LOGO_POSITIONS)}"),
    "logo_scale":               (lambda x: True,                        "0.0–1.0, e.g. 0.1"),
    "logo_opacity":             (lambda x: True,                        "0.0–1.0, e.g. 0.8"),
    "logo_margin_px":           (lambda x: x.isdigit(),                 "Margin in pixels"),
    "logo_enabled":             (lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"),
    "stop_on_error_in_playlist":(lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"),
    "reconnect_on_stream_error":(lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"),
    "quality_preset":           (lambda x: x in QUALITY_PRESETS,        f"Options: {', '.join(QUALITY_PRESETS.keys())}"),
}

INT_FIELDS = {"fps","gop_size","loop_count","reconnect_delay_seconds","max_reconnect_attempts",
              "open_timeout_seconds","read_timeout_seconds","video_thread_count","audio_sample_rate","audio_channels","logo_margin_px"}
FLOAT_FIELDS = {"logo_scale","logo_opacity"}
BOOL_FIELDS = {"logo_enabled","stop_on_error_in_playlist","reconnect_on_stream_error"}


def parse_field_value(field: str, raw: str):
    if field in BOOL_FIELDS:
        return raw.lower() in ("on", "true", "1")
    if field in INT_FIELDS:
        return int(raw)
    if field in FLOAT_FIELDS:
        return float(raw)
    return raw


async def handle_set_command(chat_id: int, text: str):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    parts = text.split(maxsplit=2)

    if len(parts) == 1:
        # /set with no args — show all settable fields
        lines = ["⚙️ <b>All configurable fields:</b>\n",
                 "Usage: <code>/set &lt;field&gt; &lt;value&gt;</code>\n"]
        for field, (_, desc) in SETTABLE_FIELDS.items():
            cur = session.get(field)
            lines.append(f"  <code>{field}</code> = <code>{esc(str(cur))}</code>\n    ↳ {esc(desc)}")
        return send_message(chat_id, "\n".join(lines), reply_markup=get_main_keyboard(session))

    field = parts[1].lower()
    if field not in SETTABLE_FIELDS:
        close = [f for f in SETTABLE_FIELDS if field in f]
        hint = f"\nDid you mean: <code>{', '.join(close)}</code>?" if close else ""
        return send_message(chat_id,
            f"❌ Unknown field <code>{esc(field)}</code>.{hint}\n"
            f"Use /set to see all fields.",
            reply_markup=get_main_keyboard(session))

    if len(parts) < 3:
        validator, desc = SETTABLE_FIELDS[field]
        cur = session.get(field)
        return send_message(chat_id,
            f"📝 <b>Field:</b> <code>{field}</code>\n"
            f"<b>Current value:</b> <code>{esc(str(cur))}</code>\n"
            f"<b>Expected:</b> {esc(desc)}\n\n"
            f"Usage: <code>/set {field} &lt;value&gt;</code>",
            reply_markup=get_main_keyboard(session))

    raw_value = parts[2].strip()
    validator, desc = SETTABLE_FIELDS[field]

    if not validator(raw_value):
        return send_message(chat_id,
            f"❌ Invalid value for <code>{field}</code>.\n"
            f"Expected: {esc(desc)}\n"
            f"Got: <code>{esc(raw_value)}</code>",
            reply_markup=get_main_keyboard(session))

    with lock:
        parsed = parse_field_value(field, raw_value)
        session[field] = parsed

        # Apply quality preset side effects
        if field == "quality_preset" and raw_value in QUALITY_PRESETS:
            for k, v in QUALITY_PRESETS[raw_value].items():
                session[k] = v

    append_user_live_log(chat_id, f"Set {field} = {parsed}")
    return send_message(chat_id,
        f"✅ <b>{field}</b> set to <code>{esc(str(parsed))}</code>",
        reply_markup=get_main_keyboard(session))


# ──────────────────────────────────────────────
# LOGO HANDLER
# ──────────────────────────────────────────────
async def handle_logo_upload(chat_id: int, message: dict):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]

    file_id = None
    filename = None
    mime_type = "image/png"

    if message.get("document"):
        doc = message["document"]
        file_id = doc.get("file_id")
        filename = doc.get("file_name", f"logo_{chat_id}.png")
        mime_type = doc.get("mime_type", "image/png")
    elif message.get("photo"):
        # Take the largest photo
        photos = message["photo"]
        file_id = photos[-1]["file_id"]
        filename = f"photo_{chat_id}.jpg"
        mime_type = "image/jpeg"

    if not file_id:
        return send_message(chat_id, "❌ Could not get file from message.", reply_markup=get_main_keyboard(session))

    # Webhook-only mode: Telegram sends only file_id, not the actual bytes.
    # Downloading requires an outbound getFile API call which is blocked on HuggingFace Spaces.
    with lock:
        session['current_step'] = None
    return send_message(chat_id,
        "⚠️ <b>Logo upload is unavailable in webhook-only mode.</b>\n\n"
        "This deployment blocks outbound requests, so the bot cannot download "
        "files from Telegram's servers.\n\n"
        "To use a logo, host the image publicly and set it via a URL workaround, "
        "or run the bot in an environment that allows outbound connections.",
        reply_markup=get_main_keyboard(session))


# ──────────────────────────────────────────────
# SCHEDULE HANDLER
# ──────────────────────────────────────────────
# Scheduling flow (conversation steps):
#   sched_when      → user picks "timer" or "datetime"
#   sched_timer     → user types "in X minutes/hours"
#   sched_datetime  → user types YYYY-MM-DD HH:MM:SS
#   sched_name      → user types a friendly job name
#   sched_rtmp      → user types the RTMP output URL for this job
#   sched_input     → user types the input URL for this job
# Then job is registered with its own output+input URLs (independent of session).

def _list_schedules(chat_id: int) -> str:
    """Build schedule list text."""
    jobs = scheduler.get_jobs()
    user_jobs = [j for j in jobs if j.id.startswith(f"stream_{chat_id}_")]
    lines = ["🕒 <b>Scheduled Streams</b> <i>(in-memory, lost on restart)</i>\n"]
    if user_jobs:
        for j in user_jobs:
            rt = j.next_run_time.strftime("%Y-%m-%d %H:%M:%S UTC") if j.next_run_time else "N/A"
            meta = _job_store.get(j.id, {})
            rtmp = meta.get("output_url", "?")
            inp  = meta.get("input_url",  "?")
            lines.append(
                f"  • <b>{esc(j.name)}</b>\n"
                f"    ⏰ <code>{rt}</code>\n"
                f"    📡 <code>{esc(rtmp[:60])}</code>\n"
                f"    🎬 <code>{esc(inp[:60])}</code>\n"
                f"    Cancel: <code>/schedule cancel {esc(j.id)}</code>"
            )
    else:
        lines.append("  No scheduled streams.")
    lines += [
        "",
        "➕ <b>Add a schedule:</b>  use <b>🕒 Schedule</b> button or <code>/schedule new</code>",
        "❌ <b>Cancel:</b>  <code>/schedule cancel &lt;job_id&gt;</code>",
    ]
    return "\n".join(lines)


def get_schedule_when_keyboard():
    return {"inline_keyboard": [
        [{"text": "⏱ Timer (in X minutes)", "callback_data": "sched_pick_timer"},
         {"text": "📅 Date & Time (UTC)",   "callback_data": "sched_pick_datetime"}],
        [{"text": "❌ Cancel",               "callback_data": "sched_cancel_setup"}],
    ]}


def get_schedule_menu_keyboard(session: dict = None):
    """Keyboard shown on the schedule list view."""
    return {"inline_keyboard": [
        [{"text": "➕ New Schedule", "callback_data": "sched_new"}],
        [{"text": "🔙 Back",         "callback_data": "settings_done"}],
    ]}


async def handle_schedule_command(chat_id: int, text: str):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    parts = text.split(maxsplit=2)
    sub = parts[1].lower() if len(parts) > 1 else ""

    # List / entry point
    if not sub or sub == "list":
        return send_message(chat_id, _list_schedules(chat_id),
                            reply_markup=get_main_keyboard(session))

    # Start new schedule conversation
    if sub == "new":
        with lock:
            session["current_step"] = "sched_when"
            session["_sched_draft"] = {}
        return send_message(chat_id,
            "🕒 <b>New Scheduled Stream</b>\n\nWhen should it start?",
            reply_markup=get_schedule_when_keyboard())

    # Cancel a job
    if sub == "cancel":
        job_id = parts[2].strip() if len(parts) > 2 else None
        if not job_id:
            return send_message(chat_id,
                "Usage: <code>/schedule cancel &lt;job_id&gt;</code>",
                reply_markup=get_main_keyboard(session))
        try:
            scheduler.remove_job(job_id)
            _job_store.pop(job_id, None)
            return send_message(chat_id,
                f"✅ Cancelled: <code>{esc(job_id)}</code>",
                reply_markup=get_main_keyboard(session))
        except Exception:
            return send_message(chat_id,
                f"❌ Job not found: <code>{esc(job_id)}</code>",
                reply_markup=get_main_keyboard(session))

    return send_message(chat_id,
        "⚠️ Unknown sub-command.\nTry /schedule, /schedule new, or /schedule cancel &lt;id&gt;",
        reply_markup=get_main_keyboard(session))


async def handle_schedule_conversation(chat_id: int, text: str):
    """Handle text input during schedule setup conversation steps."""
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    step = session.get("current_step", "")
    draft = session.setdefault("_sched_draft", {})

    if step == "sched_timer":
        # Parse "30", "30m", "2h", "1h30m", "90 minutes" etc.
        raw = text.strip().lower().replace(" ", "")
        minutes = 0
        import re as _re
        hm = _re.match(r"^(?:(\d+)h)?(?:(\d+)m?)?$", raw)
        plain = _re.match(r"^(\d+)$", raw)
        if hm and (hm.group(1) or hm.group(2)):
            minutes = int(hm.group(1) or 0) * 60 + int(hm.group(2) or 0)
        elif plain:
            minutes = int(plain.group(1))
        if minutes <= 0:
            return send_message(chat_id,
                "❌ Could not parse duration. Try <code>30</code>, <code>2h</code>, <code>1h30m</code>\nOr /cancel.")
        trigger_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=minutes)
        draft["trigger_time"] = trigger_time
        with lock:
            session["current_step"] = "sched_name"
        return send_message(chat_id,
            f"✅ Timer set: <b>{minutes} minute(s)</b> from now\n"
            f"  → <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code>\n\n"
            f"Give this schedule a <b>name</b> (or type <code>skip</code>):")

    elif step == "sched_datetime":
        raw = text.strip()
        try:
            # Accept both "YYYY-MM-DD HH:MM:SS" and "YYYY-MM-DD HH:MM"
            fmt = "%Y-%m-%d %H:%M:%S" if len(raw) > 16 else "%Y-%m-%d %H:%M"
            trigger_time = datetime.datetime.strptime(raw, fmt).replace(tzinfo=datetime.timezone.utc)
        except ValueError:
            return send_message(chat_id,
                "❌ Invalid format. Use <code>YYYY-MM-DD HH:MM:SS</code> (UTC)\nOr /cancel.")
        if trigger_time <= datetime.datetime.now(datetime.timezone.utc):
            return send_message(chat_id,
                "❌ That time is in the past. Enter a future time (UTC):\nOr /cancel.")
        draft["trigger_time"] = trigger_time
        with lock:
            session["current_step"] = "sched_name"
        return send_message(chat_id,
            f"✅ Time set: <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code>\n\n"
            f"Give this schedule a <b>name</b> (or type <code>skip</code>):")

    elif step == "sched_name":
        name = text.strip()
        draft["name"] = f"Scheduled Stream {chat_id}" if name.lower() == "skip" or not name else name
        with lock:
            session["current_step"] = "sched_rtmp"
        return send_message(chat_id,
            f"📡 <b>Step: RTMP Output URL</b>\n\n"
            f"Enter the RTMP URL to stream to:\n"
            f"<code>rtmp://a.rtmp.youtube.com/live2/YOUR_KEY</code>\n\n"
            f"Or type <code>use</code> to use your current output URL (<code>{esc(session.get('output_url', 'not set'))}</code>)\n"
            f"Or /cancel.")

    elif step == "sched_rtmp":
        raw = text.strip()
        if raw.lower() == "use":
            url = session.get("output_url", "")
        else:
            url = raw
        if not validate_url(url):
            return send_message(chat_id,
                "❌ Invalid URL. Must start with rtmp/http/https/rtsp/...\nTry again or /cancel.")
        draft["output_url"] = url
        with lock:
            session["current_step"] = "sched_input"
        return send_message(chat_id,
            f"🎬 <b>Step: Input URL</b>\n\n"
            f"Enter the stream/video input URL:\n"
            f"<code>https://example.com/video.mp4</code>\n\n"
            f"Or type <code>use</code> to use your current playlist first item (<code>{esc((session.get('input_url_playlist') or ['not set'])[0])}</code>)\n"
            f"Or /cancel.")

    elif step == "sched_input":
        raw = text.strip()
        if raw.lower() == "use":
            pl = session.get("input_url_playlist", [])
            url = pl[0] if pl else ""
        else:
            url = raw
        if not validate_url(url):
            return send_message(chat_id,
                "❌ Invalid URL. Must start with http/https/rtmp/rtsp/...\nTry again or /cancel.")
        draft["input_url"] = url

        # All info collected — register the job
        trigger_time = draft["trigger_time"]
        job_name     = draft["name"]
        output_url   = draft["output_url"]
        input_url    = draft["input_url"]

        job_id = f"stream_{chat_id}_{int(trigger_time.timestamp())}"
        try:
            scheduler.add_job(
                _scheduled_stream_runner,
                "date",
                run_date=trigger_time,
                args=[chat_id, output_url, input_url],
                id=job_id,
                name=job_name,
                replace_existing=True,
            )
            # Store metadata in our own dict (APScheduler Jobs are frozen)
            _job_store[job_id] = {"output_url": output_url, "input_url": input_url}
        except Exception as e:
            return send_message(chat_id, f"❌ Failed to register job: {esc(str(e))}",
                                reply_markup=get_main_keyboard(session))

        mins_away = int((trigger_time - datetime.datetime.now(datetime.timezone.utc)).total_seconds() / 60)
        with lock:
            session["current_step"] = None
            session["_sched_draft"] = {}

        return send_message(chat_id,
            f"✅ <b>Stream Scheduled!</b>\n\n"
            f"  📛 <b>Name:</b> {esc(job_name)}\n"
            f"  ⏰ <b>Time:</b> <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code> (~{mins_away}m from now)\n"
            f"  📡 <b>Output:</b> <code>{esc(output_url[:60])}</code>\n"
            f"  🎬 <b>Input:</b>  <code>{esc(input_url[:60])}</code>\n"
            f"  🔑 <b>ID:</b> <code>{job_id}</code>\n\n"
            f"View all: /schedule\nCancel: <code>/schedule cancel {job_id}</code>",
            reply_markup=get_main_keyboard(session))

    # Fallback — unknown step
    with lock:
        session["current_step"] = None
    return send_message(chat_id,
        "⚠️ Schedule setup cancelled (unknown step).",
        reply_markup=get_main_keyboard(session))


def _scheduled_stream_runner(chat_id: int, output_url: str = None, input_url: str = None):
    """Run by APScheduler at the scheduled time. Overrides session URLs if provided."""
    logger.info(f"[Scheduler] Firing scheduled stream for chat {chat_id}")
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]

    # Override session URLs with the per-job ones
    if output_url:
        with lock:
            session["output_url"] = output_url
            session["active_output_url"] = output_url
    if input_url:
        with lock:
            session["input_url_playlist"] = [input_url]

    # Notify user via queued outbound message (returned on next webhook)
    msg_text = (
        f"⏰ <b>Scheduled stream is starting!</b>\n\n"
        f"📡 Output: <code>{output_url or session.get('output_url', '—')}</code>\n"
        f"🎬 Input: <code>{input_url or (session.get('input_url_playlist') or ['—'])[0]}</code>\n\n"
        f"Use /abort to stop or /status for live status."
    )
    enqueue_outbound_message(chat_id, send_message(chat_id, msg_text, reply_markup=get_main_keyboard(session)))
    # Push immediate SSE notification
    push_sse_event(chat_id, {
        "type": "notification",
        "subtype": "scheduled_start",
        "text": msg_text,
        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    })

    loop = _main_event_loop
    if loop and loop.is_running():
        asyncio.run_coroutine_threadsafe(start_stream_handler(chat_id), loop)
    else:
        asyncio.run(start_stream_handler(chat_id))


# ──────────────────────────────────────────────
# HELP TEXT
# ──────────────────────────────────────────────
def get_help_text() -> str:
    return (
        f"<b>🤖 Advanced Stream Bot v{APP_VERSION}</b>\n"
        "<i>In-memory: settings lost on restart</i>\n\n"
        "<b>━━ Core Commands ━━</b>\n"
        "  /start — Home screen &amp; status\n"
        "  /stream — Start streaming\n"
        "  /pause — Pause stream\n"
        "  /resume — Resume stream\n"
        "  /abort — Stop stream\n"
        "  /status — Detailed status\n"
        "  /reboot — ⚡ Force Reboot (last resort if bot stops responding)\n\n"
        "<b>━━ Configuration ━━</b>\n"
        "  /set — List all settings\n"
        "  /set &lt;field&gt; &lt;value&gt; — Set a value\n"
        "  /settings — Interactive settings menu\n"
        "  /reset — Restore all defaults\n\n"
        "<b>━━ Playlist ━━</b>\n"
        "  /playlist show\n"
        "  /playlist add &lt;url&gt;\n"
        "  /playlist remove &lt;index|last&gt;\n"
        "  /playlist clear\n\n"
        "<b>━━ Logo ━━</b>\n"
        "  /set_logo — Upload logo (PNG/JPG)\n"
        "  /set logo_enabled on|off\n"
        "  /set logo_position top_right|...\n"
        "  /set logo_scale 0.1\n"
        "  /set logo_opacity 0.8\n\n"
        "<b>━━ Schedule ━━</b>\n"
        "  /schedule — List &amp; manage schedules\n"
        "  /schedule new — Create a new schedule (conversation)\n"
        "  /schedule cancel &lt;id&gt; — Cancel a scheduled stream\n\n"
        "<b>📅 Scheduling supports:</b>\n"
        "  ⏱ Timer: <i>in X minutes/hours</i>\n"
        "  📅 Date &amp; time: <code>YYYY-MM-DD HH:MM:SS</code> (UTC)\n"
        "  Each schedule has its own RTMP output + input URL\n\n"
        "<b>━━ Diagnostics ━━</b>\n"
        "  /logs — Recent user logs\n"
        "  /globallogs — System logs\n\n"
        "<b>━━ Emergency ━━</b>\n"
        "  /reboot — Kills all threads, resets state to idle.\n"
        "            Use if the bot stops responding or gets stuck.\n"
        "            Your URL/settings are preserved.\n\n"
        "<b>⚠️ Data is in-memory only — lost on restart.</b>"
    )


# ──────────────────────────────────────────────
# MAIN UPDATE HANDLER
# ──────────────────────────────────────────────
async def handle_telegram_update(update: dict):
    try:
        return await _handle_update_inner(update)
    except Exception as e:
        logger.error(f"Unhandled update error: {e}", exc_info=True)
        return {"status": "ok"}


async def _handle_update_inner(update: dict):
    # ── Message ──
    if "message" in update:
        msg = update["message"]
        chat_id = msg["chat"]["id"]
        session = get_user_session(chat_id)
        lock = session_locks[chat_id]
        text = msg.get("text", "")

        # File upload
        if msg.get("photo") or msg.get("document"):
            if session.get("current_step") == "awaiting_logo":
                return await handle_logo_upload(chat_id, msg)
            else:
                return send_message(chat_id,
                    "📎 Got a file — but I wasn't expecting one.\n"
                    "Use /set_logo first if you want to upload a logo.",
                    reply_markup=get_main_keyboard(session))

        if not text:
            return {"status": "ok"}

        command = text.split()[0].lower() if text.startswith('/') else None

        # ── Cancel during conversation ──
        if text.strip().lower() == "/cancel":
            with lock:
                session['current_step'] = None
                session['current_step_index'] = 0
                session['conversation_fields_list'] = []
                session['settings_editing_field'] = None
            return send_message(chat_id, "✅ Cancelled.", reply_markup=get_main_keyboard(session))

        # ── In a conversation step ──
        if session.get('current_step') and not command:
            return await handle_conversation_input(chat_id, text)

        # ── Commands ──
        if command:
            logger.info(f"[Chat {chat_id}] Command: {text[:80]}")

            if command in ("/start", "/menu", "/home"):
                with lock:
                    session['current_step'] = None
                return send_message(chat_id, compose_status_message(chat_id, include_config=True),
                                    reply_markup=get_main_keyboard(session))

            elif command == "/help":
                return send_message(chat_id, get_help_text(), reply_markup=get_main_keyboard(session))

            elif command == "/stream":
                return await start_stream_handler(chat_id)

            elif command == "/pause":
                return await pause_stream_handler(chat_id)

            elif command == "/resume":
                return await resume_stream_handler(chat_id)

            elif command == "/abort":
                return await abort_stream_handler(chat_id)

            elif command in ("/reboot", "/forcereboot"):
                return await force_reboot_handler(chat_id)

            elif command == "/status":
                return send_message(chat_id, compose_status_message(chat_id, include_config=True),
                                    reply_markup=get_main_keyboard(session))

            elif command == "/settings":
                return send_message(chat_id,
                    "⚙️ <b>Settings</b>\n\n"
                    + format_settings_display(session) + "\n\n"
                    "<b>To change a setting:</b>\n"
                    "<code>/set &lt;field&gt; &lt;value&gt;</code>\n\n"
                    "Or use the buttons below to navigate interactively.",
                    reply_markup=get_settings_keyboard(session))

            elif command == "/set":
                return await handle_set_command(chat_id, text)

            elif command == "/reset":
                return send_message(chat_id,
                    "⚠️ <b>Reset all settings to defaults?</b>\nThis cannot be undone.",
                    reply_markup=get_reset_confirm_keyboard())

            elif command == "/playlist":
                return await handle_playlist_command(chat_id, text)

            elif command == "/set_logo":
                with lock:
                    session['current_step'] = "awaiting_logo"
                return send_message(chat_id,
                    "🖼 <b>Upload Logo</b>\n\n"
                    "Send a PNG or JPG image now.\n"
                    "<i>Max 5 MB. /cancel to abort.</i>",
                    reply_markup=get_main_keyboard(session))

            elif command == "/schedule":
                return await handle_schedule_command(chat_id, text)

            elif command == "/logs":
                # Send a snapshot. User can tap 📋 Logs button for the live-updating bubble.
                logs = session.get('live_log_lines_user', [])
                log_tail = "\n".join(logs[-30:]) if logs else "No logs yet."
                return send_message(chat_id,
                    f"📋 <b>Stream Logs (last 30):</b>\n<pre>{esc(log_tail)}</pre>\n\n"
                    f"<i>💡 Tap the 📋 Logs button below for a live-updating view.</i>",
                    reply_markup=get_main_keyboard(session))

            elif command == "/globallogs":
                last = "\n".join(live_log_lines_global[-20:])
                return send_message(chat_id,
                    "📜 <b>Global Bot Logs (last 20):</b>\n<pre>" + esc(last) + "</pre>",
                    reply_markup=get_main_keyboard(session))

            else:
                # Unknown command — if in conversation, pass through
                if session.get('current_step'):
                    return await handle_conversation_input(chat_id, text)
                return send_message(chat_id,
                    f"❓ Unknown command: <code>{esc(text.split()[0])}</code>\nUse /help for available commands.",
                    reply_markup=get_main_keyboard(session))

        else:
            # Plain text — conversation input or hint
            if session.get('current_step'):
                return await handle_conversation_input(chat_id, text)
            return send_message(chat_id,
                "💬 Type a <b>command</b> or tap a button below.\n/help — see all commands.",
                reply_markup=get_main_keyboard(session))

    # ── Callback Query ──
    elif "callback_query" in update:
        cq = update["callback_query"]
        chat_id = cq["message"]["chat"]["id"]
        message_id = cq["message"]["message_id"]
        data = cq["data"]
        session = get_user_session(chat_id)
        lock = session_locks[chat_id]

        logger.info(f"[Chat {chat_id}] Callback: {data}")
        ack = answer_callback_query(cq["id"])

        # ── UX mode helpers ──────────────────────────────────────────────────
        # ux_mode "send"  → always sendMessage (new bubble, default)
        # ux_mode "edit"  → editMessageText in-place (replaces the button msg)
        ux = session.get("ux_mode", "send")

        def reply(text, kb=None):
            """Send or edit depending on ux_mode."""
            kb = kb or get_main_keyboard(session)
            if ux == "edit":
                return edit_message_text(chat_id, message_id, text, reply_markup=kb)
            return send_message(chat_id, text, reply_markup=kb)

        def edit(text, kb=None):
            """Always edit in-place (used for sub-menus that must stay in same msg)."""
            return edit_message_text(chat_id, message_id, text,
                                     reply_markup=kb or get_main_keyboard(session))

        # ─────────────────────────────────────────────────────────────────────
        # Stream controls
        if data == "stream_start":
            mid = message_id if ux == "edit" else None
            return [ack, await start_stream_handler(chat_id, message_id_to_edit=mid)]
        elif data == "stream_pause":
            mid = message_id if ux == "edit" else None
            return [ack, await pause_stream_handler(chat_id, message_id=mid)]
        elif data == "stream_resume":
            mid = message_id if ux == "edit" else None
            return [ack, await resume_stream_handler(chat_id, message_id=mid)]
        elif data == "stream_abort":
            mid = message_id if ux == "edit" else None
            return [ack, await abort_stream_handler(chat_id, message_id=mid)]

        elif data == "force_reboot":
            return [answer_callback_query(cq["id"], "⚡ Force rebooting…"),
                    await force_reboot_handler(chat_id, message_id=message_id)]

        elif data == "stream_status":
            # Register this message as the live-view bubble (status+log mode)
            # Background thread will edit it every 2s — exactly like logg.py
            register_live_view(chat_id, message_id, mode="status")
            live_text = _build_live_view_text(chat_id, "status")
            live_kb   = _get_live_view_keyboard(chat_id)
            return [ack, edit(live_text, live_kb)]

        # ── Live-view controls (Pause / Resume / Refresh / Close) ────────────
        # Directly mirrors logg.py's handle_callback() for stop_logs/resume_logs/refresh_logs
        elif data == "lv_pause":
            set_live_view_show(chat_id, False)
            # Force one immediate edit to show the "paused" state
            live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status"))
            live_kb   = _get_live_view_keyboard(chat_id)
            with _live_view_lock:
                if chat_id in _live_views:
                    _live_views[chat_id]["last_sent"] = live_text
            return [answer_callback_query(cq["id"], "⏸ Updates paused"),
                    edit(live_text, live_kb)]

        elif data == "lv_resume":
            set_live_view_show(chat_id, True)
            _do_live_view_edit(chat_id, force=True)
            live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status"))
            live_kb   = _get_live_view_keyboard(chat_id)
            return [answer_callback_query(cq["id"], "▶️ Updates resumed"),
                    edit(live_text, live_kb)]

        elif data == "lv_refresh":
            # Force immediate rebuild — same as logg.py's refresh_logs
            _do_live_view_edit(chat_id, force=True)
            live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status"))
            live_kb   = _get_live_view_keyboard(chat_id)
            with _live_view_lock:
                if chat_id in _live_views:
                    _live_views[chat_id]["last_sent"] = live_text
            return [answer_callback_query(cq["id"], "🔄 Refreshed"),
                    edit(live_text, live_kb)]

        elif data == "lv_close":
            unregister_live_view(chat_id)
            return [answer_callback_query(cq["id"], "Live view closed"),
                    edit(compose_status_message(chat_id, include_config=False),
                         get_main_keyboard(session))]

        elif data == "stream_stop_graceful":
            with lock:
                session['stop_gracefully_flag'] = True
            append_user_live_log(chat_id, "Graceful stop requested.")
            return [ack, edit("⏳ <b>Will stop after current loop finishes.</b>\n\n"
                              + compose_status_message(chat_id))]

        # Settings navigation
        elif data == "open_settings":
            return [ack, edit(
                "⚙️ <b>Settings</b>\n\n" + format_settings_display(session) + "\n\n"
                "Tap a parameter to change it, or use <code>/set &lt;field&gt; &lt;value&gt;</code>",
                get_settings_keyboard(session))]

        elif data == "toggle_ux_mode":
            with lock:
                current_ux = session.get("ux_mode", "send")
                session["ux_mode"] = "edit" if current_ux == "send" else "send"
                new_ux = session["ux_mode"]
            label = "New Message (sendMessage)" if new_ux == "send" else "Edit In-Place (editMessageText)"
            return [ack, edit(
                f"✅ <b>UX Mode switched to:</b> <code>{label}</code>\n\n"
                f"{'💬 Each button press sends a new message.' if new_ux == 'send' else '✏️ Button presses edit the existing message in-place.'}\n\n"
                + format_settings_display(session),
                get_settings_keyboard(session))]

        elif data == "settings_done":
            return [ack, reply(compose_status_message(chat_id, True))]

        elif data == "pick_quality_preset":
            return [ack, edit("🎨 <b>Choose Quality Preset:</b>", get_quality_keyboard())]

        elif data.startswith("apply_quality_"):
            q = data.replace("apply_quality_", "")
            if q in QUALITY_PRESETS:
                with lock:
                    session['quality_preset'] = q
                    for k, v in QUALITY_PRESETS[q].items():
                        session[k] = v
                return [ack, edit(
                    f"✅ Quality preset <code>{q}</code> applied.\n\n" + format_settings_display(session),
                    get_settings_keyboard(session))]

        elif data == "set_video_codec":
            return [ack, edit("🎥 <b>Choose Video Codec:</b>", get_codec_keyboard("video"))]

        elif data.startswith("set_vcodec_"):
            codec = data.replace("set_vcodec_", "")
            with lock: session['video_codec'] = codec
            return [ack, edit(f"✅ Video codec: <code>{codec}</code>", get_settings_keyboard(session))]

        elif data == "set_audio_codec":
            return [ack, edit("🔊 <b>Choose Audio Codec:</b>", get_codec_keyboard("audio"))]

        elif data.startswith("set_acodec_"):
            codec = data.replace("set_acodec_", "")
            with lock: session['audio_codec'] = codec
            return [ack, edit(f"✅ Audio codec: <code>{codec}</code>", get_settings_keyboard(session))]

        elif data == "set_ffmpeg_preset":
            return [ack, edit("⚡ <b>Choose FFmpeg Preset:</b>", get_preset_keyboard())]

        elif data.startswith("set_preset_"):
            preset = data.replace("set_preset_", "")
            with lock: session['ffmpeg_preset'] = preset
            return [ack, edit(f"✅ Preset: <code>{preset}</code>", get_settings_keyboard(session))]

        # Resolution picker
        elif data == "pick_resolution":
            return [ack, edit("📐 <b>Choose Resolution:</b>", get_resolution_keyboard())]

        elif data.startswith("set_res_"):
            val = data.replace("set_res_", "")
            if val == "custom":
                with lock:
                    session["current_step"] = "editing_field"
                    session["settings_editing_field"] = "resolution"
                return [ack, reply(
                    "📐 <b>Custom Resolution</b>\n"
                    "Current: <code>" + esc(str(session.get("resolution"))) + "</code>\n"
                    "Format: <code>WIDTHxHEIGHT</code> e.g. <code>1280x720</code> or <code>source</code>\n\n"
                    "Type the value now, or /cancel.")]
            else:
                with lock:
                    session["resolution"] = val
                label = next((l for l, v in RESOLUTION_PRESETS if v == val), val)
                return [ack, edit(
                    f"✅ Resolution set to <b>{label}</b> (<code>{val}</code>)",
                    get_settings_keyboard(session))]

        # Inline-triggered field edits (ask user to type)
        elif data in ("set_output_url", "set_fps", "set_video_bitrate",
                      "set_audio_bitrate", "set_loop_count", "set_gop_size",
                      "set_output_format", "set_reconnect_delay", "set_open_timeout"):
            field_map = {
                "set_output_url": "output_url",
                "set_fps": "fps",
                "set_video_bitrate": "video_bitrate",
                "set_audio_bitrate": "audio_bitrate",
                "set_loop_count": "loop_count",
                "set_gop_size": "gop_size",
                "set_output_format": "output_format",
                "set_reconnect_delay": "reconnect_delay_seconds",
                "set_open_timeout": "open_timeout_seconds",
            }
            field = field_map[data]
            _, desc = SETTABLE_FIELDS[field]
            cur = session.get(field)
            with lock:
                session['current_step'] = "editing_field"
                session['settings_editing_field'] = field
            return [ack, reply(
                f"📝 <b>Set {field}</b>\n"
                f"Current: <code>{esc(str(cur))}</code>\n"
                f"Expected: {esc(desc)}\n\n"
                f"Type the new value now, or /cancel to abort.")]

        elif data in ("toggle_reconnect", "toggle_stop_on_error", "toggle_verbose_log"):
            field_map = {
                "toggle_reconnect":    "reconnect_on_stream_error",
                "toggle_stop_on_error": "stop_on_error_in_playlist",
                "toggle_verbose_log":  "verbose_ffmpeg_log",
            }
            default_map = {
                "reconnect_on_stream_error": True,
                "stop_on_error_in_playlist": True,
                "verbose_ffmpeg_log":        False,
            }
            field = field_map[data]
            with lock:
                session[field] = not session.get(field, default_map[field])
                new_val = session[field]
            labels = {
                "reconnect_on_stream_error": "Auto-Reconnect",
                "stop_on_error_in_playlist": "Stop on Error",
                "verbose_ffmpeg_log":        "Verbose FFmpeg Log",
            }
            return [ack, edit(
                f"✅ <b>{labels[field]}</b> → <code>{'on' if new_val else 'off'}</code>",
                get_settings_keyboard(session))]

        # Playlist view
        elif data == "view_playlist":
            pl = session.get('input_url_playlist', [])
            if pl:
                items = "\n".join(f"  <code>{i+1}.</code> {esc(url)}" for i, url in enumerate(pl))
                msg = (f"📜 <b>Playlist ({len(pl)} items):</b>\n{items}\n\n"
                       "<b>Manage:</b>\n"
                       "  <code>/playlist add &lt;url&gt;</code>\n"
                       "  <code>/playlist remove &lt;index|last&gt;</code>\n"
                       "  <code>/playlist clear</code>")
            else:
                msg = ("📜 <b>Playlist is empty.</b>\n"
                       "<code>/playlist add &lt;url&gt;</code>")
            return [ack, reply(msg)]

        # Logo
        elif data == "cfg_logo":
            fname = session.get('logo_original_filename', 'N/A')
            enabled = session.get('logo_enabled', False)
            has_logo = bool(session.get('logo_data_bytes'))
            logo_btns = []
            if has_logo:
                logo_btns.append([{"text": ("✅ Enabled — Disable" if enabled else "❌ Disabled — Enable"),
                                    "callback_data": "toggle_logo"}])
                logo_btns.append([{"text": "📍 Change Position", "callback_data": "change_logo_pos"}])
                logo_btns.append([{"text": "🔄 Replace Logo", "callback_data": "upload_new_logo"}])
            else:
                logo_btns.append([{"text": "📤 Upload Logo", "callback_data": "upload_new_logo"}])
            logo_btns.append([{"text": "↩ Back", "callback_data": "open_settings"}])

            lines = ["🖼 <b>Logo Configuration</b>\n"]
            if has_logo:
                lines.append(f"File: <code>{esc(fname)}</code>")
                lines.append(f"Status: {'✅ Enabled' if enabled else '❌ Disabled'}")
                lines.append(f"Position: <code>{session.get('logo_position')}</code>")
                lines.append(f"Scale: <code>{session.get('logo_scale')}</code>")
                lines.append(f"Opacity: <code>{session.get('logo_opacity')}</code>\n")
                lines.append("Change scale/opacity: <code>/set logo_scale 0.15</code>")
            else:
                lines.append("No logo uploaded.")
            return [ack, edit("\n".join(lines), {"inline_keyboard": logo_btns})]

        elif data == "toggle_logo":
            with lock:
                session['logo_enabled'] = not session.get('logo_enabled', False)
                v = session['logo_enabled']
            return [ack, edit(f"🖼 Logo {'enabled ✅' if v else 'disabled ❌'}.")]

        elif data == "change_logo_pos":
            return [ack, edit("📍 <b>Choose logo position:</b>", get_logo_pos_keyboard())]

        elif data.startswith("set_logo_pos_"):
            pos = data.replace("set_logo_pos_", "")
            with lock: session['logo_position'] = pos
            return [ack, edit(f"✅ Logo position: <code>{pos}</code>")]

        elif data == "upload_new_logo":
            with lock: session['current_step'] = "awaiting_logo"
            return [ack, reply("🖼 Send a PNG or JPG image now. /cancel to abort.")]

        # Schedule
        elif data == "cfg_schedule":
            return [ack, reply(_list_schedules(chat_id),
                               get_schedule_menu_keyboard(session))]

        elif data == "sched_new":
            with lock:
                session["current_step"] = "sched_when"
                session["_sched_draft"] = {}
            return [ack, reply(
                "🕒 <b>New Scheduled Stream</b>\n\nWhen should it start?",
                get_schedule_when_keyboard())]

        elif data == "sched_pick_timer":
            with lock:
                session["current_step"] = "sched_timer"
            return [ack, reply(
                "\u23f1 <b>Timer Setup</b>\n\n"
                "How long from now? Type a duration:\n"
                "  <code>30</code> \u2192 30 minutes\n"
                "  <code>2h</code> \u2192 2 hours\n"
                "  <code>1h30m</code> \u2192 1h 30min\n\n"
                "Or /cancel.")]

        elif data == "sched_pick_datetime":
            with lock:
                session["current_step"] = "sched_datetime"
            return [ack, reply(
                "\U0001f4c5 <b>Date &amp; Time (UTC)</b>\n\n"
                "Enter the start time:\n"
                "<code>YYYY-MM-DD HH:MM:SS</code>\n"
                "Example: <code>2025-12-31 23:55:00</code>\n\n"
                "Or /cancel.")]

        elif data == "sched_cancel_setup":
            with lock:
                session["current_step"] = None
                session["_sched_draft"] = {}
            return [ack, reply("❌ Schedule setup cancelled.",
                               get_main_keyboard(session))]

        # Quick setup
        elif data == "quick_setup":
            with lock:
                session['current_step'] = "quick_output_url"
            return [ack, reply(
                "⚙️ <b>Quick Setup</b>\n\n"
                "Step 1/2: Enter your <b>RTMP Output URL</b>:\n"
                "<code>rtmp://a.rtmp.youtube.com/live2/YOUR_STREAM_KEY</code>\n\n"
                "Or /cancel to abort.")]

        # Reset
        elif data == "confirm_reset":
            return [ack, edit("⚠️ <b>Reset all settings to defaults?</b>",
                              get_reset_confirm_keyboard())]

        elif data == "do_reset":
            reset_session_settings(chat_id)
            return [ack, reply("🔄 <b>Settings restored to defaults.</b>\n\n"
                              + compose_status_message(chat_id, True))]

        # Logs — open as live-view bubble (logs mode), same concept as logg.py /logs
        elif data == "show_user_logs":
            register_live_view(chat_id, message_id, mode="logs")
            live_text = _build_live_view_text(chat_id, "logs")
            live_kb   = _get_live_view_keyboard(chat_id)
            with _live_view_lock:
                if chat_id in _live_views:
                    _live_views[chat_id]["last_sent"] = live_text
            return [ack, edit(live_text, live_kb)]

        elif data == "show_help":
            return [ack, reply(get_help_text())]

        return [ack, answer_callback_query(cq["id"], "Unknown action", show_alert=True)]

    return {"status": "ok"}


# ──────────────────────────────────────────────
# CONVERSATION HANDLER (inline field edits / quick setup)
# ──────────────────────────────────────────────
async def handle_conversation_input(chat_id: int, text: str):
    session = get_user_session(chat_id)
    lock = session_locks[chat_id]
    step = session.get('current_step')

    if step == "editing_field":
        field = session.get('settings_editing_field')
        if not field or field not in SETTABLE_FIELDS:
            with lock:
                session['current_step'] = None
            return send_message(chat_id, "⚠️ State error — cancelled.", reply_markup=get_main_keyboard(session))

        validator, desc = SETTABLE_FIELDS[field]
        raw = text.strip()
        if not validator(raw):
            return send_message(chat_id,
                f"❌ Invalid value for <code>{field}</code>.\nExpected: {esc(desc)}\nTry again or /cancel.")

        with lock:
            session[field] = parse_field_value(field, raw)
            if field == "quality_preset" and raw in QUALITY_PRESETS:
                for k, v in QUALITY_PRESETS[raw].items():
                    session[k] = v
            session['current_step'] = None
            session['settings_editing_field'] = None

        append_user_live_log(chat_id, f"Set {field} = {raw}")
        return send_message(chat_id,
            f"✅ <b>{field}</b> set to <code>{esc(raw)}</code>\n\n" + format_settings_display(session),
            reply_markup=get_settings_keyboard(session))

    elif step == "quick_output_url":
        url = text.strip()
        if not validate_url(url):
            return send_message(chat_id,
                "❌ Invalid URL. Must start with rtmp/http/https/rtsp/...\nTry again or /cancel.")
        with lock:
            session['output_url'] = url
            session['current_step'] = "quick_input_url"
        return send_message(chat_id,
            f"✅ Output URL set.\n\n"
            f"Step 2/2: Enter your <b>first Input URL</b> (stream/video URL):\n"
            f"Or /cancel to abort.")

    elif step == "quick_input_url":
        url = text.strip()
        if not validate_url(url):
            return send_message(chat_id,
                "❌ Invalid URL. Must start with http/https/rtmp/rtsp/...\nTry again or /cancel.")
        with lock:
            session['input_url_playlist'] = [url]
            session['current_step'] = None
        return send_message(chat_id,
            f"✅ <b>Quick Setup Complete!</b>\n\n"
            f"Input URL added to playlist.\n\n"
            f"You're ready to stream! Use /stream to start.\n\n"
            + compose_status_message(chat_id, True),
            reply_markup=get_main_keyboard(session))

    elif step == "awaiting_logo":
        return send_message(chat_id,
            "🖼 Please <b>send an image file</b> (PNG/JPG), not text. /cancel to abort.")

    elif step in ("sched_when", "sched_timer", "sched_datetime",
                  "sched_name", "sched_rtmp", "sched_input"):
        return await handle_schedule_conversation(chat_id, text)

    else:
        with lock:
            session['current_step'] = None
        return send_message(chat_id,
            "💬 No active setup. Use a command or button below.",
            reply_markup=get_main_keyboard(session))


# ──────────────────────────────────────────────
# FASTAPI ENDPOINTS
# ──────────────────────────────────────────────
@app.on_event("startup")
async def startup_event():
    global _main_event_loop
    _main_event_loop = asyncio.get_event_loop()
    if not scheduler.running:
        scheduler.start()
        logger.info("APScheduler started (MemoryJobStore).")
    logger.info(f"Stream Bot v{APP_VERSION} started.")
    logger.warning("All data is in-memory — lost on restart.")


@app.on_event("shutdown")
async def shutdown_event():
    if scheduler.running:
        scheduler.shutdown()
    logger.info("Stream Bot shutdown.")


@app.post("/webhook")
async def telegram_webhook_endpoint(request: Request):
    try:
        update = await request.json()

        # Identify chat_id early to drain any pending real-time status edit
        chat_id = None
        try:
            if "message" in update:
                chat_id = update["message"]["chat"]["id"]
            elif "callback_query" in update:
                chat_id = update["callback_query"]["message"]["chat"]["id"]
        except Exception:
            pass

        response_data = await handle_telegram_update(update)

        # Collect the primary response
        primary = None
        if isinstance(response_data, list):
            items = [i for i in response_data if i and isinstance(i, dict) and i.get("method")]
            for priority in ("sendMessage", "editMessageText"):
                for item in items:
                    if item.get("method") == priority:
                        primary = item
                        break
                if primary:
                    break
            if not primary:
                for item in items:
                    if item.get("method") == "answerCallbackQuery":
                        primary = item
                        break
        elif isinstance(response_data, dict):
            primary = response_data

        # Drain any background-queued outbound messages (live-view edits, scheduler
        # notifications, state-change alerts) and return the first one as the response.
        # Telegram only accepts one response per webhook call, so extras are re-queued.
        if chat_id:
            queued_msgs = pop_outbound_messages(chat_id)
            if queued_msgs:
                if primary is None or primary.get("method") == "answerCallbackQuery":
                    primary = queued_msgs[0]
                    # Re-queue remaining for future webhook calls
                    for m in queued_msgs[1:]:
                        enqueue_outbound_message(chat_id, m)
                else:
                    # Already have a real primary — re-queue all queued for next call
                    for m in queued_msgs:
                        enqueue_outbound_message(chat_id, m)

        return primary or {"status": "ok"}

    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")
    except Exception as e:
        logger.error(f"Webhook error: {e}", exc_info=True)
        return {"status": "ok"}


@app.get("/events/{chat_id}")
async def sse_events_endpoint(chat_id: int, request: Request):
    """
    Server-Sent Events endpoint for real-time stream status.
    Connect from any HTTP client:
      GET /events/{chat_id}
      Accept: text/event-stream
    Events are JSON objects with a "type" field:
      type=state   — full status snapshot (state, frames, bytes, uptime, keyboard, etc.)
      type=log     — single log line
      type=notification — background event (e.g. scheduled start)
    """
    if chat_id not in user_sessions:
        # Auto-create session so clients can connect before first message
        get_user_session(chat_id)

    q: asyncio.Queue = asyncio.Queue(maxsize=200)
    _register_sse_subscriber(chat_id, q)

    async def event_generator():
        try:
            # Send immediate snapshot on connect
            _push_state_sse(chat_id)
            # Send recent logs as a batch
            session = get_user_session(chat_id)
            recent_logs = session.get('live_log_lines_user', [])[-20:]
            if recent_logs:
                payload = json.dumps({"type": "log_batch", "lines": recent_logs})
                yield f"data: {payload}\n\n"

            while True:
                # Check client disconnected
                if await request.is_disconnected():
                    break
                try:
                    # Wait up to 3s for an event, then send a heartbeat
                    payload = await asyncio.wait_for(q.get(), timeout=3.0)
                    yield f"data: {payload}\n\n"
                except asyncio.TimeoutError:
                    # Heartbeat to keep connection alive
                    yield f": heartbeat\n\n"
        except asyncio.CancelledError:
            pass
        finally:
            _unregister_sse_subscriber(chat_id, q)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        }
    )


@app.get("/stream-log/{chat_id}")
async def sse_log_endpoint(chat_id: int, request: Request):
    """
    Dedicated SSE endpoint for live log streaming only.
    Lighter than /events — only delivers log lines.
    """
    if chat_id not in user_sessions:
        get_user_session(chat_id)

    q: asyncio.Queue = asyncio.Queue(maxsize=200)
    _register_sse_subscriber(chat_id, q)

    async def log_generator():
        try:
            # Send recent log history first
            session = get_user_session(chat_id)
            recent_logs = session.get('live_log_lines_user', [])[-50:]
            for line in recent_logs:
                payload = json.dumps({"type": "log", "line": line})
                yield f"data: {payload}\n\n"

            while True:
                if await request.is_disconnected():
                    break
                try:
                    payload = await asyncio.wait_for(q.get(), timeout=5.0)
                    # Only forward log events
                    try:
                        ev = json.loads(payload)
                        if ev.get("type") in ("log", "log_batch"):
                            yield f"data: {payload}\n\n"
                    except Exception:
                        pass
                except asyncio.TimeoutError:
                    yield f": heartbeat\n\n"
        except asyncio.CancelledError:
            pass
        finally:
            _unregister_sse_subscriber(chat_id, q)

    return StreamingResponse(
        log_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        }
    )


@app.post("/notify/{chat_id}")
async def internal_notify_endpoint(chat_id: int, request: Request):
    """
    Internal endpoint: POST a JSON event to push to SSE subscribers.
    Body: {"type": "...", ...}  (any JSON)
    Also queues a sendMessage to be returned on the next webhook response.
    """
    try:
        body = await request.json()
    except Exception:
        body = {}
    body.setdefault("ts", datetime.datetime.now(datetime.timezone.utc).isoformat())
    push_sse_event(chat_id, body)
    # If there's a "text" field, queue it as an outbound message too
    if "text" in body and chat_id in user_sessions:
        session = get_user_session(chat_id)
        enqueue_outbound_message(chat_id, send_message(chat_id, body["text"],
                                                       reply_markup=get_main_keyboard(session)))
    return {"queued": True}


@app.get("/sessions")
async def list_sessions():
    """List all active sessions and their current state."""
    result = {}
    for chat_id, session in user_sessions.items():
        result[str(chat_id)] = {
            "state": session.get("streaming_state", "idle"),
            "frames_encoded": session.get("frames_encoded", 0),
            "bytes_sent": session.get("bytes_sent", 0),
            "sse_subscribers": len(_sse_subscribers.get(chat_id, [])),
            "queued_messages": len(_outbound_message_queue.get(chat_id, [])),
        }
    return result


@app.get("/")
async def root():
    return {
        "bot": f"Advanced Stream Bot v{APP_VERSION}",
        "status": "running",
        "endpoints": {
            "webhook": "POST /webhook",
            "status": "GET /status/{chat_id}",
            "sessions": "GET /sessions",
            "sse_events": "GET /events/{chat_id}  (text/event-stream)",
            "sse_logs": "GET /stream-log/{chat_id}  (text/event-stream)",
            "notify": "POST /notify/{chat_id}",
            "health": "GET /health",
        },
        "sessions": len(user_sessions),
        "sse_connections": sum(len(v) for v in _sse_subscribers.values()),
        "scheduler_jobs": len(scheduler.get_jobs()),
    }


@app.get("/health")
async def health():
    return {
        "status": "ok",
        "version": APP_VERSION,
        "sessions": len(user_sessions),
        "scheduler_jobs": len(scheduler.get_jobs()),
        "sse_connections": sum(len(v) for v in _sse_subscribers.values()),
        "queued_messages": sum(len(v) for v in _outbound_message_queue.values()),
    }


@app.get("/status/{chat_id}")
async def get_status_endpoint(chat_id: int):
    """HTTP polling — returns current stream state as JSON for external dashboards."""
    if chat_id not in user_sessions:
        raise HTTPException(status_code=404, detail="No session for this chat_id")
    session = get_user_session(chat_id)
    uptime = 0
    if session.get("stream_start_time"):
        try:
            uptime = int((datetime.datetime.now(datetime.timezone.utc) - session["stream_start_time"]).total_seconds())
        except Exception:
            pass
    return {
        "chat_id": chat_id,
        "state": session.get("streaming_state", "idle"),
        "state_icon": STATE_EMOJI.get(session.get("streaming_state", "idle"), "❓"),
        "frames_encoded": session.get("frames_encoded", 0),
        "bytes_sent": session.get("bytes_sent", 0),
        "uptime_seconds": uptime,
        "uptime_str": get_uptime(session.get("stream_start_time")),
        "reconnect_attempt": session.get("reconnect_attempt", 0),
        "error": session.get("error_notification_user", ""),
        "active_output_url": session.get("active_output_url", ""),
        "playlist_index": session.get("current_playlist_index", 0),
        "playlist_count": len(session.get("input_url_playlist", [])),
        "sse_subscribers": len(_sse_subscribers.get(chat_id, [])),
        "queued_messages": len(_outbound_message_queue.get(chat_id, [])),
        "keyboard": get_main_keyboard(session),
        "status_text": compose_status_message(chat_id, include_config=False),
        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }


@app.get("/poll/{chat_id}")
async def poll_pending_edit(chat_id: int):
    """
    Lightweight polling endpoint — returns the pending live-view edit (if any)
    as a Telegram Bot API method dict, then clears it.

    The Telegram Mini App / web dashboard can call this every 2s and POST the
    returned payload directly to api.telegram.org/bot<TOKEN>/editMessageText to
    achieve real-time live-view updates without waiting for an incoming webhook.

    If no edit is pending, returns {"pending": false}.
    Also always includes a fresh status snapshot so callers can update their UI.
    """
    if chat_id not in user_sessions:
        raise HTTPException(status_code=404, detail="No session for this chat_id")

    # Force-build a fresh live-view edit right now
    _do_live_view_edit(chat_id, force=True)

    with _outbound_queue_lock:
        msgs = _outbound_message_queue.pop(chat_id, [])

    # Pick the most recent editMessageText (last one wins — it's the freshest)
    edit_msg = None
    for m in reversed(msgs):
        if m.get("method") == "editMessageText":
            edit_msg = m
            break
    # Re-queue any non-edit messages (sendMessage notifications etc.)
    for m in msgs:
        if m.get("method") != "editMessageText":
            enqueue_outbound_message(chat_id, m)

    session = get_user_session(chat_id)
    return {
        "pending": edit_msg is not None,
        "edit": edit_msg,
        "status": {
            "state": session.get("streaming_state", "idle"),
            "frames_encoded": session.get("frames_encoded", 0),
            "bytes_sent": session.get("bytes_sent", 0),
            "uptime_str": get_uptime(session.get("stream_start_time")),
            "logs": session.get("live_log_lines_user", [])[-15:],
        },
        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }


# ──────────────────────────────────────────────
# MAIN
# ──────────────────────────────────────────────
if __name__ == "__main__":
    import uvicorn
    logger.info(f"Starting Advanced Stream Bot v{APP_VERSION}...")
    uvicorn.run("stream_bot:app", host="0.0.0.0", port=8000, reload=False)