"""
web_app.py
──────────
FastAPI + WebSocket server for the Aara voice concierge.
CPU-only · HuggingFace Spaces

Fixes vs v1:
  • asyncio.Semaphore(3) caps concurrent inference turns
  • Per-IP connection limit (max 5) + per-session audio rate limit (max 12/min)
  • Audio payload size cap: 512 KB
  • Session stored in LRU cache (max 200, 30-min TTL) — no memory leak
  • SILENCE_STOP_MS fixed to 900ms in all HTML (was 8000ms in minimal UI)
  • ffmpeg subprocess timeout=30s
  • request_id threads through all log lines
  • Hourly cleanup task for output_audio/
  • Silero VAD results visible in status messages
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import subprocess
import threading
import time
import uuid
from collections import OrderedDict
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional, cast

import numpy as np
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from starlette.websockets import WebSocketState

from create_hotel_database import initialize_database
from voice_agent_standalone import AaraConfig, AaraVoiceAgent, SessionState

# ─────────────────────────────────────────────
# Logging
# ─────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s %(name)s: %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger("AARA.Web")

# ─────────────────────────────────────────────
# Constants
# ─────────────────────────────────────────────

OUTPUT_AUDIO_DIR = Path("output_audio")
OUTPUT_AUDIO_DIR.mkdir(parents=True, exist_ok=True)
REFERENCE_AUDIO_DIR = Path("reference_audio")
REFERENCE_AUDIO_DIR.mkdir(parents=True, exist_ok=True)

MAX_CONCURRENT_TURNS = 3          # asyncio.Semaphore value
MAX_CONNECTIONS_PER_IP = 5        # WebSocket connections per IP address
MAX_AUDIO_PER_MINUTE = 12         # Audio messages per session per minute
MAX_AUDIO_BYTES = 512 * 1024      # 512 KB — ~30s of WebM/Opus at typical bitrates
AUDIO_CLEANUP_INTERVAL = 3600     # Seconds between audio file cleanup runs
AUDIO_MAX_AGE = 3600              # Delete output_audio files older than 1 hour

# ─────────────────────────────────────────────
# Voice options for UI (English only)
# ─────────────────────────────────────────────

VOICE_OPTIONS_HTML = (
    '<option value="female">🎙 Priya / Female</option>'
    '<option value="male">🎙 Raj / Male</option>'
)


def _audio_url_from_path(path: Optional[str]) -> Optional[str]:
    if not path:
        return None
    try:
        file_path = Path(path)
        if file_path.exists():
            return f"/audio/{file_path.name}"
    except Exception:
        return None
    return None


def _reference_url_from_path(path: Optional[str]) -> Optional[str]:
    if not path:
        return None
    try:
        file_path = Path(path)
        if file_path.exists() and file_path.parent.resolve() == REFERENCE_AUDIO_DIR.resolve():
            return f"/reference-audio/{file_path.name}"
    except Exception:
        return None
    return None


def _websocket_is_open(websocket: WebSocket) -> bool:
    return (
        getattr(websocket, "client_state", None) != WebSocketState.DISCONNECTED
        and getattr(websocket, "application_state", None) != WebSocketState.DISCONNECTED
    )


async def _safe_send_json(websocket: WebSocket, payload: dict[str, Any]) -> None:
    if not _websocket_is_open(websocket):
        raise WebSocketDisconnect(code=1001)
    try:
        await websocket.send_json(payload)
    except WebSocketDisconnect:
        raise
    except RuntimeError as exc:
        if "websocket.send" in str(exc).lower():
            raise WebSocketDisconnect(code=1001) from exc
        raise


async def _safe_close(websocket: WebSocket, code: int = 1000) -> None:
    if not _websocket_is_open(websocket):
        return
    try:
        await websocket.close(code=code)
    except RuntimeError:
        return


# ═══════════════════════════════════════════════════════════════════
# Session LRU cache (replaces unbounded dict — no memory leak)
# ═══════════════════════════════════════════════════════════════════

@dataclass
class BrowserSessionState:
    session: SessionState = field(default_factory=SessionState)
    greeted: bool = False


class _SessionLRU:
    """Thread-safe LRU cache for BrowserSessionState with TTL eviction."""

    def __init__(self, maxsize: int = 200, ttl_seconds: int = 1800) -> None:
        self._cache: OrderedDict[str, tuple[float, BrowserSessionState]] = OrderedDict()
        self._maxsize = maxsize
        self._ttl = ttl_seconds
        self._lock = threading.Lock()

    def get(self, key: str) -> Optional[BrowserSessionState]:
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None
            ts, value = entry
            if time.time() - ts > self._ttl:
                del self._cache[key]
                return None
            self._cache.move_to_end(key)
            self._cache[key] = (time.time(), value)
            return value

    def set(self, key: str, value: BrowserSessionState) -> None:
        with self._lock:
            if key in self._cache:
                self._cache.move_to_end(key)
            self._cache[key] = (time.time(), value)
            while len(self._cache) > self._maxsize:
                self._cache.popitem(last=False)

    def touch(self, key: str) -> None:
        with self._lock:
            entry = self._cache.get(key)
            if entry is not None:
                self._cache.move_to_end(key)
                self._cache[key] = (time.time(), entry[1])

    def __len__(self) -> int:
        return len(self._cache)


# ═══════════════════════════════════════════════════════════════════
# Rate limiting
# ═══════════════════════════════════════════════════════════════════

_ip_connections: dict[str, int] = {}
_ip_conn_lock = threading.Lock()
_session_audio_times: dict[str, list[float]] = {}
_audio_times_lock = threading.Lock()


def _get_client_ip(websocket: WebSocket) -> str:
    client = getattr(websocket, "client", None)
    return client.host if client else "unknown"


def _ip_can_connect(ip: str) -> bool:
    with _ip_conn_lock:
        return _ip_connections.get(ip, 0) < MAX_CONNECTIONS_PER_IP


def _ip_connect(ip: str) -> None:
    with _ip_conn_lock:
        _ip_connections[ip] = _ip_connections.get(ip, 0) + 1


def _ip_disconnect(ip: str) -> None:
    with _ip_conn_lock:
        _ip_connections[ip] = max(0, _ip_connections.get(ip, 1) - 1)


def _session_is_rate_limited(session_id: str) -> bool:
    """Return True if this session has exceeded MAX_AUDIO_PER_MINUTE audio messages."""
    now = time.time()
    cutoff = now - 60.0
    with _audio_times_lock:
        times = _session_audio_times.get(session_id, [])
        times = [t for t in times if t > cutoff]
        _session_audio_times[session_id] = times
        if len(times) >= MAX_AUDIO_PER_MINUTE:
            return True
        times.append(now)
        _session_audio_times[session_id] = times
        return False


# ═══════════════════════════════════════════════════════════════════
# Global state
# ═══════════════════════════════════════════════════════════════════

_agent: Optional[AaraVoiceAgent] = None
_agent_lock = threading.Lock()
_browser_sessions = _SessionLRU(maxsize=200, ttl_seconds=1800)
_turn_semaphore: Optional[asyncio.Semaphore] = None  # created in lifespan


def get_agent() -> AaraVoiceAgent:
    global _agent
    if _agent is None:
        with _agent_lock:
            if _agent is None:
                logger.info("Starting shared Sahara Star agent ...")
                cfg = AaraConfig()
                cfg.unload_models_between_turns = False
                cfg.enable_multi_pass_asr = False
                cfg.enable_server_tts = True         # Prefer consistent server TTS, fallback stays in browser
                # NOTE: asr_confidence_threshold, llm_max_tokens etc. inherit from
                # AaraConfig defaults — do NOT override them here.
                cfg.enable_silero_vad = True         # Use package API; no torch.hub prompt needed
                if not Path(cfg.db_path).exists():
                    initialize_database(cfg.db_path, verbose=False)
                _agent = AaraVoiceAgent(cfg)
    return _agent


# ═══════════════════════════════════════════════════════════════════
# Audio cleanup task
# ═══════════════════════════════════════════════════════════════════

async def _cleanup_audio_loop() -> None:
    """Periodically delete old output_audio files to prevent disk fill."""
    while True:
        await asyncio.sleep(AUDIO_CLEANUP_INTERVAL)
        cutoff = time.time() - AUDIO_MAX_AGE
        deleted = 0
        for f in OUTPUT_AUDIO_DIR.glob("*"):
            try:
                if f.is_file() and f.stat().st_mtime < cutoff:
                    f.unlink()
                    deleted += 1
            except Exception:
                pass
        if deleted:
            logger.info("Audio cleanup: deleted %d old files", deleted)

        # Also evict old response cache entries
        try:
            agent = get_agent()
            evicted = agent.llm.cache.evict_old_audio(AUDIO_MAX_AGE)
            if evicted:
                logger.info("Cache eviction: removed %d old audio entries", evicted)
        except Exception:
            pass


# ═══════════════════════════════════════════════════════════════════
# Session helpers
# ═══════════════════════════════════════════════════════════════════

def _normalise_language(language: Optional[str]) -> str:
    """Always return English - multilingual support removed."""
    return "en"


def _restore_session(client_id: str, snapshot: Optional[dict[str, Any]]) -> SessionState:
    session = SessionState()
    session.session_id = client_id
    if not snapshot:
        return session
    session.language_preference = str(snapshot.get("language") or "en")
    session.voice_preference = str(snapshot.get("voice_preference") or "female")
    session.last_intent = str(snapshot.get("last_intent") or "")
    try:
        restored: Any = json.loads(snapshot.get("state_json") or "{}")
        if isinstance(restored, dict):
            session.task_state.update(cast(dict[str, Any], restored))
    except Exception:
        pass
    return session


def get_browser_session(client_id: str) -> BrowserSessionState:
    existing = _browser_sessions.get(client_id)
    if existing is not None:
        return existing
    agent = get_agent()
    db = cast(Any, agent.db)
    snapshot = db.get_session_snapshot(client_id)
    session = _restore_session(client_id, snapshot)
    greeted = bool(snapshot and int(snapshot.get("turn_count") or 0) > 0)
    browser_session = BrowserSessionState(session=session, greeted=greeted)
    _browser_sessions.set(client_id, browser_session)
    return browser_session


def update_preferences(browser_session: BrowserSessionState, language: Optional[str], voice_preference: Optional[str]) -> None:
    agent = get_agent()
    session = browser_session.session
    session.language_preference = "en"
    if voice_preference:
        agent.set_voice_preference_for_session(voice_preference, session)


# ═══════════════════════════════════════════════════════════════════
# Audio decode
# ═══════════════════════════════════════════════════════════════════

def _decode_browser_audio(waveform_bytes: bytes) -> np.ndarray:
    """
    Decode WebM/Opus bytes to float32 PCM at 16kHz using ffmpeg.
    FIXED: added timeout=30 to prevent hung subprocesses.
    """
    ffmpeg_cmd = [
        "ffmpeg", "-y", "-i", "pipe:0",
        "-f", "f32le", "-acodec", "pcm_f32le",
        "-ar", "16000", "-ac", "1",
        "pipe:1",
    ]
    try:
        completed = subprocess.run(
            ffmpeg_cmd,
            input=waveform_bytes,
            check=True,
            capture_output=True,
            timeout=30,          # FIXED: was no timeout — hung on malformed audio
        )
    except FileNotFoundError as exc:
        raise RuntimeError("ffmpeg is required to decode browser audio") from exc
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError("ffmpeg timed out decoding audio (malformed or too large)") from exc
    except subprocess.CalledProcessError as exc:
        stderr = exc.stderr.decode(errors="ignore") if exc.stderr else str(exc)
        raise RuntimeError(f"ffmpeg failed: {stderr}") from exc

    audio = np.frombuffer(completed.stdout, dtype=np.float32).copy()
    if audio.size == 0:
        raise RuntimeError("ffmpeg produced no audio samples")
    return audio


# ═══════════════════════════════════════════════════════════════════
# Turn processing
# ═══════════════════════════════════════════════════════════════════

def process_browser_audio(
    browser_session: BrowserSessionState,
    audio_bytes: bytes,
    request_id: str = "",
) -> dict[str, Any]:
    agent = get_agent()
    logger.info(
        "[%s] Audio recv: session=%s lang=%s voice=%s bytes=%d",
        request_id,
        browser_session.session.session_id,
        browser_session.session.language_preference,
        browser_session.session.voice_preference,
        len(audio_bytes),
    )
    try:
        audio = _decode_browser_audio(audio_bytes)
    except RuntimeError as exc:
        logger.warning("[%s] Audio decode failed: %s", request_id, exc)
        return {"error": str(exc)}

    language_hint = browser_session.session.language_preference or None

    result = agent.process_turn(
        audio=audio,
        language_hint=language_hint,
        play_audio=False,
        session=browser_session.session,
        voice_preference=browser_session.session.voice_preference,
        unload_models_between_turns=agent.cfg.unload_models_between_turns,
        request_id=request_id,
    )

    if result.get("error"):
        return {"error": result["error"], **result}

    return {
        "type": "result",
        "transcription": "",
        "response": "",
        "spoken_text": result.get("response", ""),
        "language": result.get("language", browser_session.session.language_preference),
        "rtt_seconds": result.get("rtt_seconds", 0.0),
        "audio_url": _audio_url_from_path(cast(Optional[str], result.get("audio_path"))),
        "voice_profile": result.get("voice_profile", browser_session.session.voice_preference),
        "voice_reference_url": _reference_url_from_path(cast(Optional[str], result.get("voice_reference_path"))),
    }


# ═══════════════════════════════════════════════════════════════════
# WebSocket message handlers
# ═══════════════════════════════════════════════════════════════════

async def send_greeting(websocket: WebSocket, browser_session: BrowserSessionState) -> None:
    agent = get_agent()
    session = browser_session.session
    greeting_text = agent.llm.generate_greeting(
        language=session.language_preference,
        voice_preference=session.voice_preference,
        session=session,
    )
    voice_profile, audio_path = await asyncio.to_thread(
        agent.build_voice_output,
        greeting_text,
        preference=session.voice_preference,
        session=session,
        request_id=session.session_id[-8:],
    )
    session.task_state["opening_greeting_sent"] = True
    logger.info("Greeting: session=%s text=%r", session.session_id, greeting_text[:80])
    await _safe_send_json(websocket, {
        "type": "greeting",
        "message": greeting_text,
        "spoken_text": greeting_text,
        "language": session.language_preference,
        "audio_url": _audio_url_from_path(audio_path),
        "voice_profile": voice_profile.key,
        "voice_reference_url": _reference_url_from_path(voice_profile.reference_audio_path),
    })


async def send_ready(websocket: WebSocket, client_id: str, browser_session: BrowserSessionState) -> None:
    await _safe_send_json(websocket, {
        "type": "ready",
        "message": "Connected to Sahara Star concierge.",
        "client_id": client_id,
        "session_id": browser_session.session.session_id,
    })


async def handle_settings(websocket: WebSocket, browser_session: BrowserSessionState, data: dict[str, Any]) -> None:
    update_preferences(browser_session, data.get("language"), data.get("voice_preference"))
    await _safe_send_json(websocket, {"type": "status", "message": "Settings updated.", "level": "ok"})


async def handle_reset(websocket: WebSocket, browser_session: BrowserSessionState) -> None:
    browser_session.session.clear()
    browser_session.greeted = False
    await _safe_send_json(websocket, {"type": "status", "message": "Conversation reset.", "level": "ok"})


async def handle_text_message(
    websocket: WebSocket, browser_session: BrowserSessionState, text_payload: str
) -> None:
    try:
        data = json.loads(text_payload)
    except json.JSONDecodeError:
        return
    message_type = data.get("type")
    if message_type == "start":
        # Client clicked the Start button — send greeting once, then client begins listening loop
        if not browser_session.greeted:
            await send_greeting(websocket, browser_session)
            browser_session.greeted = True
        else:
            await _safe_send_json(websocket, {"type": "status", "message": "Ready to listen.", "level": "ok"})
    elif message_type == "settings":
        await handle_settings(websocket, browser_session, data)
    elif message_type == "reset":
        await handle_reset(websocket, browser_session)


async def handle_audio_message(
    websocket: WebSocket,
    browser_session: BrowserSessionState,
    audio_bytes: bytes,
    request_id: str = "",
) -> None:
    assert _turn_semaphore is not None

    # Warn immediately if semaphore is full so the user isn't silently waiting
    if _turn_semaphore.locked() if hasattr(_turn_semaphore, "locked") else False:
        await _safe_send_json(websocket, {
            "type": "status",
            "message": "One moment, I'm helping another guest...",
            "level": "warn",
        })

    async with _turn_semaphore:
        await _safe_send_json(websocket, {
            "type": "status", "message": "Processing your request...", "level": "busy"
        })
        try:
            result = await asyncio.to_thread(
                process_browser_audio, browser_session, bytes(audio_bytes), request_id
            )
        except Exception as exc:
            logger.error("[%s] Audio processing error: %s", request_id, exc, exc_info=True)
            await _safe_send_json(websocket, {
                "type": "error", "message": "I couldn't process that audio. Please try again."
            })
            return

        if result.get("error"):
            await _safe_send_json(websocket, {"type": "error", "message": result["error"]})
            return

        await _safe_send_json(websocket, result)


async def bootstrap_session(websocket: WebSocket) -> tuple[str, BrowserSessionState]:
    try:
        hello_raw = await asyncio.wait_for(websocket.receive_text(), timeout=5.0)
        hello = json.loads(hello_raw)
    except asyncio.TimeoutError:
        hello = {"type": "hello", "client_id": str(uuid.uuid4()), "language": "en", "voice_preference": "female"}
    except Exception:
        hello = {"type": "hello", "client_id": str(uuid.uuid4()), "language": "en", "voice_preference": "female"}

    if hello.get("type") != "hello":
        await _safe_send_json(websocket, {"type": "error", "message": "Expected hello handshake."})
        await _safe_close(websocket, code=1002)
        raise WebSocketDisconnect(code=1002)

    client_id = str(hello.get("client_id") or uuid.uuid4())
    browser_session = get_browser_session(client_id)
    update_preferences(browser_session, hello.get("language"), hello.get("voice_preference"))
    await send_ready(websocket, client_id, browser_session)
    return client_id, browser_session


# ═══════════════════════════════════════════════════════════════════
# FastAPI lifespan
# ═══════════════════════════════════════════════════════════════════

@asynccontextmanager
async def lifespan(_: FastAPI):
    global _turn_semaphore
    _turn_semaphore = asyncio.Semaphore(MAX_CONCURRENT_TURNS)

    # Warm up models in background thread
    agent = await asyncio.to_thread(get_agent)
    warmup_jobs: list[tuple[str, Any]] = [
        ("ASR", asyncio.to_thread(agent.asr.warmup)),
        ("LLM", asyncio.to_thread(agent.llm.warmup)),
    ]
    if agent.tts is not None:
        warmup_jobs.append(("TTS", asyncio.to_thread(agent.tts.warmup)))
    warmup_results = await asyncio.gather(*(job for _, job in warmup_jobs), return_exceptions=True)
    for (label, _), outcome in zip(warmup_jobs, warmup_results):
        if isinstance(outcome, Exception):
            logger.warning("%s warmup issue (will retry on first request): %s", label, outcome)
        else:
            logger.info("%s warmup done", label)

    # Start background cleanup task
    cleanup_task = asyncio.create_task(_cleanup_audio_loop())

    yield

    cleanup_task.cancel()
    try:
        await cleanup_task
    except asyncio.CancelledError:
        pass


# ═══════════════════════════════════════════════════════════════════
# App
# ═══════════════════════════════════════════════════════════════════

app = FastAPI(title="Sahara Star Voice Concierge", lifespan=lifespan)
app.mount("/audio", StaticFiles(directory=str(OUTPUT_AUDIO_DIR)), name="audio")
app.mount("/reference-audio", StaticFiles(directory=str(REFERENCE_AUDIO_DIR)), name="reference-audio")


@app.get("/api/health")
async def api_health() -> JSONResponse:
    agent = get_agent()
    ok, checks = agent.health_check()
    checks["active_sessions"] = len(_browser_sessions)
    checks["semaphore_available"] = _turn_semaphore._value if _turn_semaphore else 0
    return JSONResponse({"ok": bool(ok), "checks": {str(k): bool(v) if isinstance(v, bool) else v for k, v in checks.items()}})


# ═══════════════════════════════════════════════════════════════════
# Routes
# ═══════════════════════════════════════════════════════════════════

@app.get("/", response_class=HTMLResponse)
async def index() -> HTMLResponse:
    """Single-button voice concierge UI. Each page refresh = fresh session."""
    return HTMLResponse("""<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8"/>
  <meta name="viewport" content="width=device-width, initial-scale=1"/>
  <title>Sahara Star · Voice Concierge</title>
  <link rel="preconnect" href="https://fonts.googleapis.com">
  <link href="https://fonts.googleapis.com/css2?family=Fraunces:opsz,wght@9..144,600;9..144,700&family=Manrope:wght@400;500;600;700;800&display=swap" rel="stylesheet">
  <style>
    :root{--bg:#07111f;--gold:#d8b06a;--gold2:#f2d8a7;--rose:#ff6a79;--emerald:#3dd6a4;--text:#eef4ff;--muted:#9da9bd}
    *{box-sizing:border-box;margin:0;padding:0}
    html,body{width:100%;height:100%;background:
      radial-gradient(circle at 20% 15%,rgba(216,176,106,0.18),transparent 32%),
      radial-gradient(circle at 80% 85%,rgba(61,214,164,0.10),transparent 28%),
      linear-gradient(180deg,#050a12 0%,var(--bg) 40%,#06101d 100%);
      color:var(--text);font-family:'Manrope',Arial,sans-serif;
      display:flex;flex-direction:column;align-items:center;justify-content:center;gap:0}
    /* ── Card ── */
    .card{display:flex;flex-direction:column;align-items:center;gap:28px;
      background:rgba(14,23,38,0.72);backdrop-filter:blur(20px);
      border:1px solid rgba(255,255,255,0.10);border-radius:36px;
      padding:48px 40px 44px;width:min(480px,calc(100vw - 32px));
      box-shadow:0 28px 80px rgba(0,0,0,0.40)}
    /* ── Hotel branding ── */
    .eyebrow{text-transform:uppercase;letter-spacing:0.22em;font-size:0.68rem;color:var(--gold2);opacity:.85}
    h1{font-family:'Fraunces',serif;font-size:clamp(2rem,7vw,3rem);
       line-height:0.95;letter-spacing:-0.04em;text-align:center}
    h1 span{color:var(--gold)}
    .tagline{color:var(--muted);font-size:0.88rem;text-align:center;line-height:1.6;max-width:30ch}
    /* ── Status badge ── */
    .status-row{display:flex;align-items:center;gap:9px;min-height:1.6em}
    .dot{width:9px;height:9px;border-radius:50%;flex-shrink:0;
         background:var(--muted);box-shadow:0 0 0 5px rgba(157,169,189,0.12);
         transition:background .3s,box-shadow .3s}
    .dot.ok{background:var(--emerald);box-shadow:0 0 0 6px rgba(61,214,164,0.14)}
    .dot.busy{background:var(--gold);box-shadow:0 0 0 6px rgba(216,176,106,0.18);animation:pulse-dot 1s infinite}
    .dot.warn{background:var(--rose);box-shadow:0 0 0 6px rgba(255,106,121,0.14)}
    @keyframes pulse-dot{0%,100%{transform:scale(1)}50%{transform:scale(1.25)}}
    #status{font-size:0.88rem;color:var(--muted);transition:color .25s}
    #status.ok{color:var(--text)}
    #status.busy{color:#9be7ff}
    #status.warn{color:#ffd28a}
    /* ── Mic orb ── */
    .orb-wrap{position:relative;width:136px;height:136px;display:flex;align-items:center;justify-content:center}
    .orb-ring{position:absolute;border-radius:50%;border:1.5px solid rgba(216,176,106,0.20);
              width:100%;height:100%;pointer-events:none;transition:opacity .3s}
    .orb-ring.active{opacity:1;animation:orb-expand 1.4s ease-out infinite}
    @keyframes orb-expand{0%{transform:scale(1);opacity:.45}100%{transform:scale(1.65);opacity:0}}
    #btn{position:relative;z-index:1;width:112px;height:112px;border-radius:50%;border:0;
         cursor:pointer;font:inherit;font-weight:800;font-size:0.85rem;line-height:1.3;
         color:#0b1321;background:linear-gradient(145deg,var(--gold2),var(--gold));
         box-shadow:0 12px 36px rgba(216,176,106,0.30);
         transition:transform .18s,box-shadow .18s,background .25s,color .25s;
         display:flex;flex-direction:column;align-items:center;justify-content:center;gap:5px}
    #btn:disabled{opacity:.55;cursor:not-allowed}
    #btn.listening{color:#fff;background:linear-gradient(145deg,#ff8b98,var(--rose));
      box-shadow:0 12px 36px rgba(255,106,121,0.34)}
    #btn.speaking{color:#fff;background:linear-gradient(145deg,#7ee8c9,var(--emerald));
      box-shadow:0 12px 36px rgba(61,214,164,0.30)}
    #btn:not(:disabled):hover{transform:scale(1.05)}
    .btn-icon{font-size:1.6rem;line-height:1}
    /* ── Voice-only mode ── */
    .transcript{display:none}
    .msg{padding:10px 14px;border-radius:18px;font-size:0.88rem;line-height:1.6;
         border:1px solid rgba(255,255,255,0.07);white-space:pre-wrap;word-break:break-word;
         animation:rise .22s ease}
    @keyframes rise{from{opacity:0;transform:translateY(5px)}to{opacity:1;transform:none}}
    .msg.user{background:rgba(216,176,106,0.12);border-bottom-right-radius:5px;margin-left:auto;max-width:85%;text-align:right}
    .msg.agent{background:rgba(255,255,255,0.05);border-bottom-left-radius:5px;max-width:90%}
    .msg .who{font-size:0.72rem;text-transform:uppercase;letter-spacing:0.10em;opacity:.55;margin-bottom:4px}
  </style>
</head>
<body>
  <div class="card">
    <div class="eyebrow">Sahara Star Hotels · Mumbai</div>
    <h1>Aara<br><span>Concierge</span></h1>
    <p class="tagline">Your voice-first hotel assistant.<br>Speak naturally — I understand you.</p>

    <div class="status-row">
      <div class="dot" id="dot"></div>
      <span id="status">Connecting...</span>
    </div>

    <div class="orb-wrap">
      <div class="orb-ring" id="ring"></div>
      <button id="btn" type="button" disabled>
        <span class="btn-icon">🎙️</span>
        <span id="btn-label">Start</span>
      </button>
    </div>

    <div class="transcript" id="transcript"></div>

    <div style="display:flex;gap:12px;align-items:center;width:100%;margin-top:-8px">
      <select id="voiceSel" style="flex:1;background:rgba(255,255,255,0.07);color:var(--text);
        border:1px solid rgba(255,255,255,0.15);border-radius:12px;padding:8px 12px;
        font-family:inherit;font-size:0.82rem;cursor:pointer">
        <option value="female">🎙 Priya / Female</option>
        <option value="male">🎙 Raj / Male</option>
      </select>
      <button id="applyVoice" type="button" style="background:rgba(216,176,106,0.18);
        color:var(--gold2);border:1px solid rgba(216,176,106,0.30);border-radius:12px;
        padding:8px 14px;font-family:inherit;font-size:0.82rem;cursor:pointer;white-space:nowrap">
        Apply
      </button>
      <button id="previewVoice" type="button" style="background:rgba(61,214,164,0.12);
        color:#b9ffe8;border:1px solid rgba(61,214,164,0.28);border-radius:12px;
        padding:8px 14px;font-family:inherit;font-size:0.82rem;cursor:pointer;white-space:nowrap">
        Preview
      </button>
    </div>
  </div>

  <script>
  (function(){
    var btn   = document.getElementById("btn");
    var label = document.getElementById("btn-label");
    var dot   = document.getElementById("dot");
    var st    = document.getElementById("status");
    var ring  = document.getElementById("ring");
    var chat  = document.getElementById("transcript");
    var voiceSel    = document.getElementById("voiceSel");
    var applyVoice  = document.getElementById("applyVoice");
    var previewVoice = document.getElementById("previewVoice");

    // ── Fresh UUID every refresh = new session every time ──────────
    var clientId;
    try { clientId = crypto.randomUUID(); }
    catch(e) { clientId = Date.now().toString(16) + "-" + Math.random().toString(16).slice(2); }

    // ── State: CONNECTING | READY | LISTENING | SENDING | SPEAKING | STOPPED
    var STATE = "CONNECTING";
    var socket = null, recorder = null, stream = null, chunks = [];
    var actx = null, asrc = null, analyser = null, rafId = null;
    var speechOk = false, lastVoiceAt = 0, recAt = 0, firstSpeechAt = 0, noiseFloor = 0.003;
    var voices = [];
    var player = null;

    var SILENCE_MS = 900, MAX_REC_MS = 12000, NO_SPEECH_MS = 2800, MIN_SPEECH_MS = 320;
    var BASE_RMS = 0.008, SPEECH_MULT = 2.4, HOLD_MULT = 1.55;

    // English only (all other languages removed)
    var LOCALES = {en:"en-IN"};
    var VOICE_REFERENCE = {
      female: "/reference-audio/female_reference.wav",
      male: "/reference-audio/male_reference.wav"
    };

    // ── UI helpers ─────────────────────────────────────────────────
    function uiSync() {
      var S = STATE;
      var modeMap = {CONNECTING:"busy",READY:"ok",LISTENING:"busy",SENDING:"busy",SPEAKING:"ok",STOPPED:"ok"};
      var lblMap  = {CONNECTING:"Wait…",READY:"Start",LISTENING:"Listening",SENDING:"Sending…",SPEAKING:"Speaking",STOPPED:"Restart"};
      dot.className   = "dot " + (modeMap[S] || "busy");
      ring.className  = "orb-ring" + (S === "LISTENING" ? " active" : "");
      btn.disabled    = (S === "CONNECTING" || S === "SENDING" || S === "SPEAKING");
      btn.className   = S === "LISTENING" ? "listening" : S === "SPEAKING" ? "speaking" : "";
      label.textContent = lblMap[S] || "Start";
      st.className    = modeMap[S] || "busy";
    }
    function setStatus(msg, mode) {
      st.textContent = msg;
      st.className   = mode || "ok";
      dot.className  = "dot " + (mode || "ok");
    }
    function goTo(s) { STATE = s; uiSync(); }

    function addMsg(role, text, who) { return; }

    function currentVoicePreference() {
      return voiceSel ? (voiceSel.value || "female") : "female";
    }

    // ── Audio cleanup ──────────────────────────────────────────────
    function cleanAudio() {
      if (rafId) { cancelAnimationFrame(rafId); rafId = null; }
      try { if (asrc) asrc.disconnect(); } catch(e) {}
      try { if (analyser) analyser.disconnect(); } catch(e) {}
      try { if (actx) { var c = actx; actx = null; c.close(); } } catch(e) {}
      asrc = null; analyser = null; speechOk = false; lastVoiceAt = 0; recAt = 0;
      firstSpeechAt = 0; noiseFloor = 0.003;
    }
    function stopStream() {
      if (stream) { stream.getTracks().forEach(function(t){t.stop();}); stream = null; }
    }
    function stopPlayback() {
      if (window.speechSynthesis) window.speechSynthesis.cancel();
      if (player) {
        try { player.pause(); player.currentTime = 0; } catch(e) {}
      }
      player = null;
    }

    // ── Silence detector ───────────────────────────────────────────
    function monSilence() {
      if (STATE !== "LISTENING" || !analyser) return;
      var buf = new Uint8Array(analyser.fftSize);
      analyser.getByteTimeDomainData(buf);
      var ss = 0;
      for (var i = 0; i < buf.length; i++) { var s = (buf[i] - 128) / 128; ss += s * s; }
      var rms = Math.sqrt(ss / buf.length), now = Date.now();
      noiseFloor = Math.min(0.02, Math.max(0.0025, noiseFloor * 0.92 + rms * 0.08));
      var speechThreshold = Math.max(BASE_RMS, noiseFloor * SPEECH_MULT);
      var holdThreshold = Math.max(BASE_RMS * 0.75, noiseFloor * HOLD_MULT);
      if (rms >= speechThreshold || (speechOk && rms >= holdThreshold)) {
        speechOk = true;
        if (!firstSpeechAt) firstSpeechAt = now;
        lastVoiceAt = now;
      }
      if (!speechOk && now - recAt >= NO_SPEECH_MS) { commitRec(true); return; }
      if (speechOk && now - lastVoiceAt >= SILENCE_MS) { commitRec(false); return; }
      if (now - recAt >= MAX_REC_MS) { commitRec(!firstSpeechAt || (now - firstSpeechAt < MIN_SPEECH_MS)); return; }
      rafId = requestAnimationFrame(monSilence);
    }

    // ── TTS ────────────────────────────────────────────────────────
    function refreshVoices() { voices = window.speechSynthesis ? window.speechSynthesis.getVoices() : []; }
    function pickVoice(lang, preference) {
      var locale = (LOCALES[lang] || "en-IN").toLowerCase();
      var prefix = locale.split("-")[0];
      var pref = (preference || currentVoicePreference()).toLowerCase();
      var wanted = pref === "male"
        ? ["male","raj","prabhat","david","guy","mark","ravi","man"]
        : ["female","priya","neerja","zira","aria","heera","samantha","woman"];
      var blocked = pref === "male"
        ? ["female","priya","neerja","zira","aria","heera","samantha","woman"]
        : ["male","raj","prabhat","david","guy","mark","ravi","man"];
      var best = null, bestScore = -9999;
      for (var i = 0; i < voices.length; i++) {
        var v = voices[i];
        var vlang = (v.lang || "").toLowerCase();
        var ident = ((v.name || "") + " " + (v.voiceURI || "")).toLowerCase();
        var score = -20;
        if (vlang === locale) score += 30;
        else if (vlang.indexOf(prefix) === 0) score += 18;
        else if (!vlang) score += 0;
        else continue;
        for (var j = 0; j < wanted.length; j++) {
          if (ident.indexOf(wanted[j]) >= 0) score += 8;
        }
        for (var k = 0; k < blocked.length; k++) {
          if (ident.indexOf(blocked[k]) >= 0) score -= 5;
        }
        if (score > bestScore) {
          bestScore = score;
          best = v;
        }
      }
      return best || voices[0] || null;
    }
    function speak(text, lang, preference, endState) {
      var nextState = endState || "LISTENING";
      if (!text || !window.speechSynthesis) { goTo(nextState); return; }
      stopPlayback();
      var u = new SpeechSynthesisUtterance(text);
      var v = pickVoice(lang || "en", preference);
      if (v) { u.voice = v; u.lang = v.lang || (LOCALES[lang || "en"] || "en-IN"); }
      else    { u.lang = LOCALES[lang || "en"] || "en-IN"; }
      u.rate   = 1.0;
      u.onstart = function() { goTo("SPEAKING"); };
      u.onend   = function() { if (STATE !== "STOPPED") goTo(nextState); };
      u.onerror = function() { if (STATE !== "STOPPED") goTo(nextState); };
      window.speechSynthesis.speak(u);
    }
    function playAudioUrl(url, fallbackText, lang, preference, endState) {
      var nextState = endState || "LISTENING";
      if (!url) { speak(fallbackText, lang, preference, nextState); return; }
      stopPlayback();
      try {
        player = new Audio(url + (url.indexOf("?") >= 0 ? "&" : "?") + "ts=" + Date.now());
        player.onplay = function() { goTo("SPEAKING"); };
        player.onended = function() { player = null; if (STATE !== "STOPPED") goTo(nextState); };
        player.onerror = function() {
          player = null;
          if (STATE !== "STOPPED") speak(fallbackText, lang, preference, nextState);
        };
        var played = player.play();
        if (played && played.catch) {
          played.catch(function() {
            player = null;
            speak(fallbackText, lang, preference, nextState);
          });
        }
      } catch(e) {
        player = null;
        speak(fallbackText, lang, preference, nextState);
      }
    }
    function playReferenceVoice() {
      var pref = currentVoicePreference();
      var refUrl = VOICE_REFERENCE[pref];
      if (!refUrl) return;
      var restoreState = STATE === "STOPPED" ? "STOPPED" : "READY";
      setStatus("Playing " + (pref === "male" ? "Raj" : "Priya") + " reference clip…", "ok");
      playAudioUrl(refUrl, "", "en", pref, restoreState);
    }

    // ── Recording ──────────────────────────────────────────────────
    function startRec() {
      if (STATE !== "LISTENING" || recorder || stream) return;
      stopPlayback();
      navigator.mediaDevices.getUserMedia({audio:{
        echoCancellation:true, noiseSuppression:true, autoGainControl:true, channelCount:1
      }}).then(function(s) {
        stream = s; chunks = [];
        var opts = MediaRecorder.isTypeSupported("audio/webm;codecs=opus") ? {mimeType:"audio/webm;codecs=opus"} : {};
        recorder = new MediaRecorder(stream, opts);
        recorder.ondataavailable = function(e) { if (e.data && e.data.size > 0) chunks.push(e.data); };
        recorder.onstop = function() {
          var had = !!firstSpeechAt;
          cleanAudio(); stopStream(); recorder = null;
          if (!had || chunks.length === 0) {
            chunks = [];
            if (STATE === "SENDING") goTo("LISTENING");
            setStatus("No speech — listening again…", "warn");
            return;
          }
          var blob = new Blob(chunks, {type:"audio/webm"}); chunks = [];
          if (!socket || socket.readyState !== WebSocket.OPEN) { goTo("READY"); return; }
          goTo("SENDING");
          setStatus("Sending your question…", "busy");
          socket.send(blob);
        };
        recorder.start();
        recAt = Date.now(); lastVoiceAt = Date.now(); firstSpeechAt = 0; noiseFloor = 0.003; speechOk = false;
        try {
          var AC = window.AudioContext || window.webkitAudioContext;
          if (AC) {
            actx = new AC();
            if (actx.state === "suspended") actx.resume();
            asrc    = actx.createMediaStreamSource(stream);
            analyser = actx.createAnalyser(); analyser.fftSize = 2048;
            asrc.connect(analyser);
            rafId = requestAnimationFrame(monSilence);
          }
        } catch(e) { cleanAudio(); }
        setStatus("Listening — speak now…", "busy");
      }).catch(function(e) {
        stopStream(); cleanAudio(); recorder = null;
        setStatus(e.name === "NotAllowedError" ? "Microphone denied." : "Mic error.", "warn");
        goTo("STOPPED");
      });
    }

    function commitRec(discard) {
      if (!recorder || recorder.state === "inactive") return;
      if (discard) { cleanAudio(); chunks = []; recorder.stop(); recorder = null; stopStream(); return; }
      recorder.stop();
    }

    // ── Ticker: drives LISTENING state without relying on TTS onend ─
    setInterval(function() {
      if (STATE === "LISTENING" && !recorder && socket && socket.readyState === WebSocket.OPEN) {
        startRec();
      }
      // Unstick SPEAKING if synthesis already stopped
      if (STATE === "SPEAKING" && !player && window.speechSynthesis && !window.speechSynthesis.speaking) {
        goTo("LISTENING");
      }
    }, 400);

    // ── WebSocket ──────────────────────────────────────────────────
    function connect() {
      goTo("CONNECTING");
      setStatus("Connecting…", "busy");
      var proto = location.protocol === "https:" ? "wss:" : "ws:";
      var wsUrl = proto + "//" + location.host + "/ws";
      socket = new WebSocket(wsUrl);

      socket.onopen = function() {
          socket.send(JSON.stringify({
            type:"hello", client_id:clientId, language:"en",
            voice_preference: voiceSel ? voiceSel.value : "female"
          }));
        goTo("READY");
        setStatus("Connected — press Start to begin.", "ok");
      };

      socket.onmessage = function(event) {
        var p;
        try { p = JSON.parse(event.data); } catch(e) { return; }

        if (p.type === "ready") { return; }

        if (p.type === "greeting") {
          if (p.voice_profile && p.voice_reference_url) { VOICE_REFERENCE[p.voice_profile] = p.voice_reference_url; }
          addMsg("agent", p.spoken_text || p.message || "Welcome!", "Aara");
          setStatus("Aara is speaking…", "busy");
          goTo("SPEAKING");
          playAudioUrl(
            p.audio_url,
            p.spoken_text || p.message,
            p.language || "en",
            p.voice_profile || currentVoicePreference(),
            "LISTENING"
          );
          return;
        }

        if (p.type === "status") {
          setStatus(p.message || "Working…", p.level || "busy"); return;
        }

        if (p.type === "result") {
          if (p.voice_profile && p.voice_reference_url) { VOICE_REFERENCE[p.voice_profile] = p.voice_reference_url; }
          if (p.transcription) addMsg("user", p.transcription, "You said");
          var reply = p.spoken_text || p.response || "";
          if (p.response) addMsg("agent", p.response, "Aara");
          if (reply) {
            setStatus("Aara is responding…", "busy");
            goTo("SPEAKING");
            playAudioUrl(
              p.audio_url,
              reply,
              p.language || "en",
              p.voice_profile || currentVoicePreference(),
              "LISTENING"
            );
          } else {
            goTo("LISTENING");
          }
          return;
        }

        if (p.type === "error") {
          var msg = p.message || "Sorry, could not understand. Please try again.";
          addMsg("agent", msg, "Aara");
          setStatus(msg, "warn");
          setTimeout(function() { if (STATE !== "STOPPED") goTo("LISTENING"); }, 1500);
          return;
        }
      };

      socket.onclose = function() {
        if (STATE !== "STOPPED") goTo("CONNECTING");
        setStatus("Reconnecting…", "warn");
        setTimeout(connect, 1500);
      };
      socket.onerror = function() { setStatus("Connection error. Retrying…", "warn"); };
    }

    // ── Button ─────────────────────────────────────────────────────
    btn.addEventListener("click", function() {
      if (STATE === "CONNECTING") return;

      if (STATE === "READY" || STATE === "STOPPED") {
        // Send start to trigger greeting, then listening begins
        if (socket && socket.readyState === WebSocket.OPEN) {
          socket.send(JSON.stringify({type:"start", client_id:clientId}));
          setStatus("Starting…", "busy");
        }
        return;
      }

      // Any active state → stop
      commitRec(true);
      cleanAudio(); stopStream(); recorder = null;
      stopPlayback();
      goTo("STOPPED");
      setStatus("Stopped. Press Restart to continue.", "ok");
    });

    // ── Voice Apply button ─────────────────────────────────────────
    if (applyVoice) {
      applyVoice.addEventListener("click", function() {
        if (socket && socket.readyState === WebSocket.OPEN) {
          socket.send(JSON.stringify({
            type: "settings",
            voice_preference: currentVoicePreference()
          }));
          setStatus("Voice updated to " + (voiceSel ? voiceSel.options[voiceSel.selectedIndex].text : "Voice") + ".", "ok");
        }
      });
    }
    if (previewVoice) {
      previewVoice.addEventListener("click", function() {
        if (recorder) { commitRec(true); }
        cleanAudio(); stopStream(); recorder = null;
        playReferenceVoice();
      });
    }

    // ── Boot ───────────────────────────────────────────────────────
    if (window.speechSynthesis) {
      refreshVoices();
      window.speechSynthesis.onvoiceschanged = refreshVoices;
    }
    connect();
    uiSync();
  })();
  </script>
</body>
</html>""")


# ═══════════════════════════════════════════════════════════════════
# WebSocket endpoint
# ═══════════════════════════════════════════════════════════════════

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    client_ip = _get_client_ip(websocket)
    client_id: Optional[str] = None
    browser_session: Optional[BrowserSessionState] = None

    # Per-IP connection limit
    if not _ip_can_connect(client_ip):
        logger.warning("Connection rejected (IP limit): %s", client_ip)
        await _safe_send_json(websocket, {"type": "error", "message": "Too many connections from your address. Please try again shortly."})
        await _safe_close(websocket, code=1013)
        return

    _ip_connect(client_ip)

    try:
        logger.info("WS accepted from %s", client_ip)
        client_id, browser_session = await bootstrap_session(websocket)

        # Greeting is sent lazily when client sends {type:"start"} (button click)
        # so a page refresh always starts fresh without an auto-spoken greeting.

        while True:
            message = await websocket.receive()
            if message.get("type") == "websocket.disconnect":
                break

            text_payload: Optional[str] = message.get("text")
            bytes_payload: Optional[bytes] = message.get("bytes")

            if text_payload and browser_session:
                await handle_text_message(websocket, browser_session, text_payload)
                continue

            if bytes_payload and browser_session:
                # Audio payload size guard
                if len(bytes_payload) > MAX_AUDIO_BYTES:
                    logger.warning("[%s] Audio rejected: %d bytes > %d limit", client_id, len(bytes_payload), MAX_AUDIO_BYTES)
                    await _safe_send_json(websocket, {"type": "error", "message": f"Audio too large ({len(bytes_payload)//1024}KB). Please keep requests under 30 seconds."})
                    continue

                # Per-session rate limit
                if _session_is_rate_limited(browser_session.session.session_id):
                    logger.warning("[%s] Rate limited", client_id)
                    await _safe_send_json(websocket, {"type": "error", "message": "Too many requests. Please wait a moment before speaking again."})
                    continue

                request_id = uuid.uuid4().hex[:8]
                await handle_audio_message(websocket, browser_session, bytes(bytes_payload), request_id=request_id)
                _browser_sessions.touch(client_id)  # Refresh TTL on activity

    except WebSocketDisconnect:
        logger.info("WS disconnected: %s (%s)", client_id, client_ip)
    except Exception as exc:
        logger.error("WS failure: %s", exc, exc_info=True)
        try:
            await _safe_send_json(websocket, {"type": "error", "message": "An internal error occurred."})
        except Exception:
            pass
    finally:
        _ip_disconnect(client_ip)
        if client_id and browser_session is not None:
            _browser_sessions.set(client_id, browser_session)


if __name__ == "__main__":
    import uvicorn

    port = int(os.environ.get("PORT", "7860"))
    host = "0.0.0.0" if os.environ.get("SPACE_ID") else "127.0.0.1"
    uvicorn.run("web_app:app", host=host, port=port, reload=False)