app/services/session_store.py

"""
session_store.py
────────────────
Async, fire-and-forget JSON-based chat session storage.
Each browser session gets its own JSON file in data/sessions/.

Rules:
  • NEVER raises exceptions that propagate upward — any failure is silently logged.
  • ALL disk I/O is done with asyncio to avoid blocking the event loop.
  • Thread-safe: uses an asyncio.Lock per session_id to avoid race conditions.
"""

import asyncio
import json
import logging
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict

from app.core.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()

# ── Directory where session JSON files are stored ───────────────────────────
def _get_sessions_dir() -> Path:
    """Returns the current sessions directory from settings."""
    return settings.sessions_dir


# ── Per-session locks to prevent concurrent write corruption ─────────────────
_session_locks: Dict[str, asyncio.Lock] = {}
_locks_meta_lock = asyncio.Lock()  # protects the _session_locks dict itself


def _get_session_path(session_id: str) -> Path:
    """Returns the absolute path for a session's JSON file."""
    return _get_sessions_dir() / f"{session_id}.json"


def _now_iso() -> str:
    """Returns the current UTC time as an ISO-8601 string."""
    return datetime.now(timezone.utc).isoformat()


async def _get_lock(session_id: str) -> asyncio.Lock:
    """Gets or creates a per-session asyncio.Lock in a thread-safe manner."""
    async with _locks_meta_lock:
        if session_id not in _session_locks:
            _session_locks[session_id] = asyncio.Lock()
        return _session_locks[session_id]


def _ensure_sessions_dir() -> None:
    """Creates the sessions directory if it does not exist. Sync, called once."""
    try:
        _get_sessions_dir().mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        logger.error("session_store: cannot create sessions dir: %s", exc)


# Call once at import time so the directory is always ready.
_ensure_sessions_dir()


async def ensure_session(session_id: str) -> None:
    """
    Creates the session JSON file if it doesn't exist yet.
    Safe to call on every page load — does nothing if file exists.
    """
    path = _get_session_path(session_id)
    if path.exists():
        return
    lock = await _get_lock(session_id)
    async with lock:
        # Double-check after acquiring lock
        if path.exists():
            return
        skeleton = {
            "session_id": session_id,
            "user_name": "default name",
            "created_at": _now_iso(),
            "last_active": _now_iso(),
            "message_count": 0,
            "current_chat_id": 1,
            "messages": [],
            "status": "AI",
            "blocked": False,
            "rating": None,
            "alert_id": 0,
            "alert_type": "",
            "archived": False,
            "archived_chats": [],
            "sla_metrics": {},
        }
        try:
            await asyncio.to_thread(_write_json, path, skeleton)
            logger.info("session_store: created session %s", session_id)
        except Exception as exc:
            logger.error("session_store: failed to create session %s: %s", session_id, exc)


async def save_message(session_id: str, question: str, answer: str, chat_id: int = None) -> None:
    """
    Appends a Question+Answer pair with timestamps to the session's JSON file.
    Fire-and-forget safe — caller should wrap in asyncio.create_task().

    Args:
        chat_id: If provided, message is saved to this specific chat thread.
                 If None, uses current_chat_id from the session file.
    """
    path = _get_session_path(session_id)
    logger.info("session_store: save_message called for session_id=%s, path=%s", session_id, path)
    lock = await _get_lock(session_id)
    async with lock:
        try:
            # Load current data
            if path.exists():
                data = await asyncio.to_thread(_read_json, path)
            else:
                # Session file missing — recreate it gracefully
                data = {
                    "session_id": session_id,
                    "user_name": "default name",
                    "created_at": _now_iso(),
                    "last_active": _now_iso(),
                    "message_count": 0,
                    "current_chat_id": 1,
                    "messages": [],
                    "status": "AI",
                    "blocked": False,
                    "rating": None,
                    "alert_id": 0,
                    "alert_type": "",
                    "archived": False,
                    "archived_chats": [],
                    "sla_metrics": {},
                }

            timestamp = _now_iso()
            # Use explicit chat_id if provided, otherwise fall back to current
            if chat_id is None:
                chat_id = data.get("current_chat_id", 1)

            # --- SLA Agent Response Tracking ---
            if data.get("status") == "HUMAN_TAKEOVER" and not question and answer:
                if "sla_metrics" not in data:
                    data["sla_metrics"] = {}
                if "agent_reply_times" not in data["sla_metrics"]:
                    data["sla_metrics"]["agent_reply_times"] = []
                
                # Find the last message from the user in this chat.
                # Skip if the last message was already an admin reply (no question),
                # which means the admin is sending a follow-up without a new user
                # question in between — we shouldn't double-count that.
                last_user_msg = None
                for m in reversed(data["messages"]):
                    if m.get("chat_id", 1) != chat_id:
                        continue
                    if m.get("question"):
                        last_user_msg = m
                        break
                    if not m.get("question") and m.get("answer"):
                        # Last message in this chat was an admin reply — skip tracking
                        break
                
                if last_user_msg:
                    try:
                        u_time = datetime.fromisoformat(last_user_msg["timestamp"].replace("Z", "+00:00"))
                        a_time = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
                        delta_sec = (a_time - u_time).total_seconds()
                        if delta_sec > 0:
                            data["sla_metrics"]["agent_reply_times"].append(delta_sec)
                    except Exception as e:
                        logger.error("Error calculating agent response time: %s", e)
            # -----------------------------------

            data["messages"].append({
                "id": data["message_count"] + 1,
                "chat_id": chat_id,
                "timestamp": timestamp,
                "question": question,
                "answer": answer,
            })
            data["message_count"] += 1
            data["last_active"] = timestamp

            await asyncio.to_thread(_write_json, path, data)
            logger.info("session_store: successfully saved message to %s", path)
        except Exception as exc:
            logger.error(
                "session_store: failed to save message for session %s: %s",
                session_id,
                exc,
            )


async def get_all_sessions(search: str = "", filter_type: str = "", skip: int = 0, limit: int = 50) -> list:
    """
    Returns a list of all session summaries (metadata only, no full messages).
    Used by the Admin Panel sidebar. Supports pagination, searching, and filtering.
    """
    try:
        files = await asyncio.to_thread(_list_json_files)
        sessions = []
        
        now_utc = datetime.now(timezone.utc)
        today_date = now_utc.date()
        search_lower = search.lower() if search else ""
        
        for fpath in files:
            try:
                data = await asyncio.to_thread(_read_json, fpath)
                s_id = data.get("session_id", fpath.stem)
                u_name = data.get("user_name", "")
                
                if search_lower:
                    if search_lower not in str(s_id).lower() and search_lower not in str(u_name).lower():
                        continue
                
                status = data.get("status", "AI")
                blocked = data.get("blocked", False)
                archived = data.get("archived", False)
                archived_chats = data.get("archived_chats") or []
                last_active = data.get("last_active", "")
                
                if filter_type == "archived":
                    if not (archived or len(archived_chats) > 0):
                        continue
                elif filter_type == "blocked":
                    if not blocked:
                        continue
                elif filter_type == "queued":
                    if status != "QUEUED" or archived or blocked:
                        continue
                elif filter_type == "daily":
                    if blocked or archived:
                        continue
                    created_at = data.get("created_at", "")
                    if not created_at:
                        continue
                    try:
                        dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
                        if dt.astimezone(timezone.utc).date() != today_date:
                            continue
                    except Exception:
                        continue
                elif filter_type == "sessions" or not filter_type:
                    if blocked or archived:
                        continue
                
                sessions.append({
                    "session_id": s_id,
                    "user_name": u_name,
                    "created_at": data.get("created_at", ""),
                    "last_active": last_active,
                    "message_count": data.get("message_count", 0),
                    "current_chat_id": data.get("current_chat_id", 1),
                    "status": status,
                    "blocked": blocked,
                    "rating": data.get("rating", None),
                    "alert_id": data.get("alert_id", 0),
                    "alert_type": data.get("alert_type", ""),
                    "archived": archived,
                    "archived_chats": archived_chats,
                })
            except Exception as exc:
                logger.warning("session_store: could not read %s: %s", fpath, exc)
        
        # Sort newest first
        sessions.sort(key=lambda s: s.get("last_active", ""), reverse=True)
        return sessions[skip:skip+limit]
    except Exception as exc:
        logger.error("session_store: get_all_sessions failed: %s", exc)
        return []

async def get_metrics() -> dict:
    """Calculates global metrics efficiently for the dashboard."""
    try:
        files = await asyncio.to_thread(_list_json_files)
        live = 0
        queued = 0
        daily = 0
        blocked_count = 0
        
        now_utc = datetime.now(timezone.utc)
        today_date = now_utc.date()
        
        for fpath in files:
            try:
                data = await asyncio.to_thread(_read_json, fpath)
                
                blocked = data.get("blocked", False)
                archived = data.get("archived", False)
                status = data.get("status", "AI")
                last_active = data.get("last_active", "")
                
                if blocked:
                    blocked_count += 1
                
                if not blocked and not archived:
                    if status == "QUEUED":
                        queued += 1
                    elif status != "ENDED":
                        live += 1
                    
                    created_at = data.get("created_at", "")
                    if created_at:
                        try:
                            dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
                            if dt.astimezone(timezone.utc).date() == today_date:
                                daily += 1
                        except Exception:
                            pass
            except Exception:
                pass
                
        return {
            "live": live,
            "queued": queued,
            "daily": daily,
            "blocked": blocked_count
        }
    except Exception as exc:
        logger.error("session_store: get_metrics failed: %s", exc)
        return {"live": 0, "queued": 0, "daily": 0, "blocked": 0}


async def get_session(session_id: str) -> dict:
    """
    Returns the full session data (including all messages) for the Admin Panel.
    Returns an empty dict if the session does not exist.
    """
    path = _get_session_path(session_id)
    try:
        if not path.exists():
            return {}
        return await asyncio.to_thread(_read_json, path)
    except Exception as exc:
        logger.error("session_store: get_session(%s) failed: %s", session_id, exc)
        return {}


async def delete_session(session_id: str) -> bool:
    """
    Deletes a session file. Returns True on success, False on failure.
    Used by Admin Panel if needed in the future.
    """
    path = _get_session_path(session_id)
    lock = await _get_lock(session_id)
    async with lock:
        try:
            if path.exists():
                await asyncio.to_thread(os.remove, path)
            return True
        except Exception as exc:
            logger.error("session_store: delete_session(%s) failed: %s", session_id, exc)
            return False


async def update_session_name(session_id: str, user_name: str) -> None:
    """Updates the user_name field in the session JSON file."""
    path = _get_session_path(session_id)
    lock = await _get_lock(session_id)
    async with lock:
        try:
            if path.exists():
                data = await asyncio.to_thread(_read_json, path)
                data["user_name"] = user_name
                await asyncio.to_thread(_write_json, path, data)
                logger.info("session_store: updated name for session %s to %s", session_id, user_name)
            else:
                # If session doesn't exist yet, we can't update it, but we could create it?
                # Usually ensure_session is called first.
                logger.warning("session_store: attempt to update name for non-existent session %s", session_id)
        except Exception as exc:
            logger.error("session_store: failed to update name for session %s: %s", session_id, exc)


async def rename_session(old_id: str, new_name: str) -> str:
    """
    Renames/merges a temp session into the canonical named session for new_name.

    Strategy:
    - If a session file for `new_name` already exists (same user logging in from
      a different browser or device), we MERGE the temp session's messages into
      that existing session and discard the temp file.
    - If no named session exists yet, we simply rename the temp file.
    - If the session is already correctly named, we just update user_name in-place.

    Always returns the canonical session_id that the caller should use.
    """
    import re
    # 1. Sanitize the name for use as a filename
    clean_name = re.sub(r'[^\w\s-]', '', new_name).strip().replace(' ', '_')
    if not clean_name:
        clean_name = "User"

    new_id = clean_name
    old_path = _get_session_path(old_id)
    new_path = _get_session_path(new_id)

    # ── Case A: session file is already correctly named ──────────────────────
    if new_id == old_id:
        update_lock = await _get_lock(old_id)
        async with update_lock:
            try:
                if old_path.exists():
                    data = await asyncio.to_thread(_read_json, old_path)
                    if data.get("user_name") != new_name:
                        data["user_name"] = new_name
                        await asyncio.to_thread(_write_json, old_path, data)
                        logger.info(
                            "session_store: updated user_name in-place for %s → '%s'",
                            old_id, new_name
                        )
            except Exception as exc:
                logger.error(
                    "session_store: failed to update user_name for %s: %s", old_id, exc
                )
        return old_id

    # ── Case B: canonical named session already exists — MERGE ───────────────
    if new_path.exists():
        lock_old = await _get_lock(old_id)
        lock_new = await _get_lock(new_id)
        async with lock_old:
            async with lock_new:
                try:
                    existing_data = await asyncio.to_thread(_read_json, new_path)

                    # Merge messages from the temp session (if any) into the
                    # existing canonical session so no history is lost.
                    if old_path.exists():
                        temp_data = await asyncio.to_thread(_read_json, old_path)
                        new_messages = temp_data.get("messages", [])
                        if new_messages:
                            existing_data["messages"].extend(new_messages)
                            existing_data["message_count"] = len(existing_data["messages"])
                            existing_data["last_active"] = _now_iso()
                            await asyncio.to_thread(_write_json, new_path, existing_data)
                            logger.info(
                                "session_store: merged %d message(s) from temp session '%s' into '%s'",
                                len(new_messages), old_id, new_id
                            )
                        # Remove the now-redundant temp file
                        await asyncio.to_thread(os.remove, old_path)
                        logger.info(
                            "session_store: removed temp session file '%s' after merge", old_id
                        )
                    else:
                        logger.info(
                            "session_store: temp session '%s' had no file; reusing '%s'",
                            old_id, new_id
                        )
                except Exception as exc:
                    logger.error(
                        "session_store: failed to merge session %s -> %s: %s", old_id, new_id, exc
                    )
        return new_id

    # ── Case C: no existing named session — simple rename ────────────────────
    lock_old = await _get_lock(old_id)
    lock_new = await _get_lock(new_id)

    async with lock_old:
        async with lock_new:
            try:
                if old_path.exists():
                    data = await asyncio.to_thread(_read_json, old_path)
                    data["session_id"] = new_id
                    data["user_name"] = new_name

                    await asyncio.to_thread(_write_json, new_path, data)
                    await asyncio.to_thread(os.remove, old_path)

                    logger.info("session_store: renamed session %s -> %s", old_id, new_id)
                    return new_id
                else:
                    logger.info(
                        "session_store: old session %s did not exist; creating new named session %s", old_id, new_id
                    )
                    skeleton = {
                        "session_id": new_id,
                        "user_name": new_name,
                        "created_at": _now_iso(),
                        "last_active": _now_iso(),
                        "message_count": 0,
                        "messages": [],
                        "status": "AI",
                        "blocked": False,
                        "rating": None,
                        "alert_id": 0,
                        "alert_type": "",
                        "archived": False,
                        "archived_chats": [],
                        "sla_metrics": {},
                    }
                    await asyncio.to_thread(_write_json, new_path, skeleton)
                    return new_id
            except Exception as exc:
                logger.error("session_store: failed to rename session %s: %s", old_id, exc)
                return old_id


# ── Sync helper functions (run via asyncio.to_thread) ────────────────────────

def _write_json(path: Path, data: dict) -> None:
    """Atomically writes JSON to a file using a temp file + rename pattern."""
    tmp_path = path.with_suffix(".tmp")
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    os.replace(tmp_path, path)  # Atomic on POSIX, best-effort on Windows


def _read_json(path: Path) -> dict:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def _list_json_files() -> list:
    return list(_get_sessions_dir().glob("*.json"))


async def update_session_field(session_id: str, field: str, value) -> bool:
    """Generic helper to update a single root field in the session JSON."""
    path = _get_session_path(session_id)
    lock = await _get_lock(session_id)
    async with lock:
        try:
            if not path.exists():
                return False
            data = await asyncio.to_thread(_read_json, path)
            
            if field == "status":
                if "sla_metrics" not in data:
                    data["sla_metrics"] = {}
                
                # If changing to QUEUED, record queued_at
                if value == "QUEUED" and data.get("status") != "QUEUED":
                    data["sla_metrics"]["queued_at"] = _now_iso()
                    # Clear old takeover/ended to prep for new escalation
                    data["sla_metrics"].pop("takeover_at", None)
                    data["sla_metrics"].pop("ended_at", None)
                
                # If changing to HUMAN_TAKEOVER, record takeover_at and calculate wait time
                elif value == "HUMAN_TAKEOVER" and data.get("status") != "HUMAN_TAKEOVER":
                    takeover_time = _now_iso()
                    data["sla_metrics"]["takeover_at"] = takeover_time
                    
                    queued_at = data["sla_metrics"].get("queued_at")
                    if queued_at:
                        try:
                            q_time = datetime.fromisoformat(queued_at.replace("Z", "+00:00"))
                            t_time = datetime.fromisoformat(takeover_time.replace("Z", "+00:00"))
                            wait_sec = (t_time - q_time).total_seconds()
                            if wait_sec >= 0:
                                data["sla_metrics"].setdefault("queue_wait_times", []).append(wait_sec)
                        except Exception:
                            pass
                
                # If leaving human-support phase, record ended_at and calculate resolution time
                elif data.get("status") in ["HUMAN_TAKEOVER", "QUEUED"] and value not in ["HUMAN_TAKEOVER", "QUEUED"]:
                    ended_time = _now_iso()
                    data["sla_metrics"]["ended_at"] = ended_time
                    
                    start_str = data["sla_metrics"].get("takeover_at") or data["sla_metrics"].get("queued_at")
                    if start_str:
                        try:
                            s_time = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
                            e_time = datetime.fromisoformat(ended_time.replace("Z", "+00:00"))
                            delta = (e_time - s_time).total_seconds()
                            if delta >= 0:
                                data["sla_metrics"].setdefault("resolution_times", []).append(delta)
                        except Exception:
                            pass
                
                # Catch-all for explicitly ended (if not handled above)
                elif value == "ENDED" and data.get("status") != "ENDED":
                    data["sla_metrics"]["ended_at"] = _now_iso()

            data[field] = value
            await asyncio.to_thread(_write_json, path, data)
            logger.info("session_store: updated %s for session %s", field, session_id)
            return True
        except Exception as exc:
            logger.error("session_store: failed to update %s for session %s: %s", field, session_id, exc)
            return False


async def archive_chat_thread(session_id: str, chat_id: int, archive: bool) -> bool:
    """Adds or removes a chat_id from the archived_chats list."""
    path = _get_session_path(session_id)
    lock = await _get_lock(session_id)
    async with lock:
        try:
            if not path.exists():
                return False
            data = await asyncio.to_thread(_read_json, path)
            archived_chats = data.get("archived_chats", [])
            
            if archive:
                if chat_id not in archived_chats:
                    archived_chats.append(chat_id)
            else:
                if chat_id in archived_chats:
                    archived_chats.remove(chat_id)
                    
            data["archived_chats"] = archived_chats
            await asyncio.to_thread(_write_json, path, data)
            logger.info("session_store: updated archived_chats for session %s: %s", session_id, archived_chats)
            return True
        except Exception as exc:
            logger.error("session_store: failed to archive chat thread %s for session %s: %s", chat_id, session_id, exc)
            return False


async def get_sla_metrics() -> dict:
    """Calculates SLA metrics across all sessions."""
    try:
        files = await asyncio.to_thread(_list_json_files)
        total_queue_time = 0
        queued_count = 0
        
        total_resolution_time = 0
        resolved_count = 0
        
        total_agent_reply_time = 0
        agent_reply_count = 0
        
        takeover_count = 0
        total_ai_sessions = 0
        
        breach_count = 0
        SLA_THRESHOLD_SEC = 3 * 60 # 3 minutes

        for fpath in files:
            try:
                data = await asyncio.to_thread(_read_json, fpath)
                
                if data.get("message_count", 0) > 0:
                    total_ai_sessions += 1

                sla = data.get("sla_metrics", {})
                
                if "takeover_at" in sla or data.get("status") == "HUMAN_TAKEOVER":
                    takeover_count += 1
                
                # Read from arrays first (new logic)
                for w in sla.get("queue_wait_times", []):
                    total_queue_time += w
                    queued_count += 1
                    if w > SLA_THRESHOLD_SEC:
                        breach_count += 1
                        
                for r in sla.get("resolution_times", []):
                    total_resolution_time += r
                    resolved_count += 1
                
                # Fallback for legacy data (before arrays were added)
                if not sla.get("queue_wait_times") and not sla.get("resolution_times"):
                    queued_at_str = sla.get("queued_at")
                    takeover_at_str = sla.get("takeover_at")
                    ended_at_str = sla.get("ended_at")
                    
                    if queued_at_str and takeover_at_str:
                        try:
                            q_time = datetime.fromisoformat(queued_at_str.replace("Z", "+00:00"))
                            t_time = datetime.fromisoformat(takeover_at_str.replace("Z", "+00:00"))
                            wait_sec = (t_time - q_time).total_seconds()
                            if wait_sec >= 0:
                                total_queue_time += wait_sec
                                queued_count += 1
                                if wait_sec > SLA_THRESHOLD_SEC:
                                    breach_count += 1
                        except Exception:
                            pass
                            
                    start_str = takeover_at_str or queued_at_str
                    if ended_at_str and start_str:
                        try:
                            s_time = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
                            e_time = datetime.fromisoformat(ended_at_str.replace("Z", "+00:00"))
                            delta = (e_time - s_time).total_seconds()
                            if delta >= 0:
                                total_resolution_time += delta
                                resolved_count += 1
                        except Exception:
                            pass
                
                reply_times = sla.get("agent_reply_times", [])
                for rt in reply_times:
                    total_agent_reply_time += rt
                    agent_reply_count += 1

            except Exception:
                pass
                
        avg_queue_time = (total_queue_time / queued_count) if queued_count > 0 else 0
        avg_resolution_time = (total_resolution_time / resolved_count) if resolved_count > 0 else 0
        avg_agent_reply_time = (total_agent_reply_time / agent_reply_count) if agent_reply_count > 0 else 0
        takeover_rate = (takeover_count / total_ai_sessions * 100) if total_ai_sessions > 0 else 0
        breach_rate = (breach_count / queued_count * 100) if queued_count > 0 else 0

        return {
            "avg_queue_time_sec": avg_queue_time,
            "avg_resolution_time_sec": avg_resolution_time,
            "avg_agent_reply_time_sec": avg_agent_reply_time,
            "takeover_rate_pct": takeover_rate,
            "breach_count": breach_count,
            "breach_rate_pct": breach_rate,
            "total_ai_sessions": total_ai_sessions,
            "queued_count": queued_count
        }
    except Exception as exc:
        logger.error("session_store: get_sla_metrics failed: %s", exc)
        return {
            "avg_queue_time_sec": 0, "avg_resolution_time_sec": 0, 
            "avg_agent_reply_time_sec": 0, "takeover_rate_pct": 0,
            "breach_count": 0, "breach_rate_pct": 0,
            "total_ai_sessions": 0, "queued_count": 0
        }