"""Session model and in-memory session store."""

import secrets
import uuid
from abc import ABC, abstractmethod
from datetime import datetime, timezone

from pydantic import BaseModel, Field, PrivateAttr

from navi.llm.base import Message


class Session(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    profile_id: str
    user_id: str | None = None  # owner; null for legacy single-user sessions
    messages: list[Message] = Field(default_factory=list)   # full display history (never compressed)
    context: list[Message] = Field(default_factory=list)    # LLM context (may be compressed)
    context_token_count: int = 0   # accumulated total; reset to 0 after compression
    pinned: bool = False
    special: bool = False  # service sessions (e.g. Synapse reactions) — hidden from the default list
    name: str | None = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    last_active: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    planning_logs: list[dict] = Field(default_factory=list)  # raw planning phase outputs per turn
    db_message_count: int = Field(default=0, exclude=True)  # messages already persisted in DB
    db_next_sequence: int = Field(default=0, exclude=True)  # next global sequence_number for this session
    archive_threshold: int = Field(default=0, exclude=True)  # seq < threshold are archived
    session_metadata: dict = Field(default_factory=dict, exclude=True)  # ephemeral session-level metadata (e.g. terminal configs)
    # List views only: the display-message count and the sidebar snippet, filled
    # without loading the messages themselves (see sessions/rows.py).
    display_message_count: int = Field(default=0, exclude=True)
    preview: str = Field(default="", exclude=True)
    # save() refreshes only rows whose mutable state changed; this is the
    # in-memory baseline it compares against (never serialized).
    _saved_fingerprints: dict = PrivateAttr(default_factory=dict)

    @property
    def display_count(self) -> int:
        """How many display messages the session has, cheaply when possible.

        List views fill ``display_message_count`` from an aggregate and leave
        ``messages`` empty; a store that hands back a hydrated session fills
        ``messages`` instead. Either answer is the same number.
        """
        return self.display_message_count or len(self.messages)


def match_snippet(content: str, search: str) -> str:
    """The sidebar snippet: ~30 chars either side of the first occurrence.

    Falls back to the head of the content when the term cannot be located by a
    plain lowercase scan (SQL matching and Python differ on a few exotic
    case-folding pairs, and a hit is still a hit).
    """
    idx = content.lower().find(search.lower())
    if idx < 0:
        return content[:60] + ("…" if len(content) > 60 else "")
    start = max(0, idx - 30)
    end = min(len(content), idx + len(search) + 30)
    snippet = content[start:end]
    if start > 0:
        snippet = "…" + snippet
    if end < len(content):
        snippet = snippet + "…"
    return snippet


def match_info_from_messages(messages: list[Message], search: str) -> tuple[list[int], str]:
    """Message positions containing ``search``, plus a preview of the first hit.

    Indices are positions in the session's *display* history — the numbering
    ``GET /sessions`` returns and clients scroll to. A store that can answer
    from SQL (``search_matches``) produces the same two values without loading
    messages.
    """
    q = search.lower()
    indices: list[int] = []
    preview = ""
    for i, msg in enumerate(messages):
        content = msg.content or ""
        if q in content.lower():
            indices.append(i)
            if not preview:
                preview = match_snippet(content, search)
    return indices, preview


class SessionStore(ABC):
    @abstractmethod
    async def create(self, profile_id: str, user_id: str | None = None) -> Session: ...

    @abstractmethod
    async def get(self, session_id: str) -> Session | None: ...

    @abstractmethod
    async def save(self, session: Session) -> None: ...

    @abstractmethod
    async def list_all(
        self,
        user_id: str | None = None,
        is_admin: bool = False,
        special: bool | None = None,
    ) -> list[Session]:
        """special: None — no filtering, False — regular only, True — special only."""

    @abstractmethod
    async def list_page(
        self,
        *,
        limit: int,
        offset: int,
        profile_id: str | None = None,
        user_id: str | None = None,
        is_admin: bool = False,
        special: bool | None = None,
    ) -> list[Session]: ...

    async def list_all_summaries(
        self,
        user_id: str | None = None,
        is_admin: bool = False,
        special: bool | None = None,
    ) -> list[Session]:
        """``list_all`` for list views: counts and previews, no message payload.

        Optional because it is an optimization, not a capability: a store that
        holds its sessions in memory answers with ``list_all`` and loses
        nothing. A database-backed store overrides it and reads aggregates.
        """
        return await self.list_all(user_id=user_id, is_admin=is_admin, special=special)

    async def search_matches(
        self, session_ids: list[str], search: str
    ) -> dict[str, tuple[list[int], str]]:
        """``{session_id: (display indices, snippet)}`` for a search result page.

        Default: no match information, which clients treat as "cannot say
        where" and fall back to showing the session without a jump target.
        """
        return {}

    @abstractmethod
    async def count_all(
        self,
        *,
        user_id: str | None = None,
        is_admin: bool = False,
        search: str | None = None,
        special: bool | None = None,
    ) -> int: ...

    @abstractmethod
    async def search_list(
        self,
        *,
        limit: int,
        offset: int,
        user_id: str | None = None,
        is_admin: bool = False,
        search: str | None = None,
        sort_by: str = "last_active",
        sort_order: str = "desc",
        special: bool | None = None,
    ) -> list[Session]: ...

    @abstractmethod
    async def delete(self, session_id: str) -> bool: ...

    @abstractmethod
    async def set_pinned(self, session_id: str, pinned: bool) -> bool: ...

    @abstractmethod
    async def set_name(self, session_id: str, name: str) -> bool: ...

    @abstractmethod
    async def set_profile(self, session_id: str, profile_id: str) -> bool:
        """Repoint a session at another profile without loading or saving it.

        A whole-session load-and-save from inside a running turn (switch_profile)
        would make the caller a second writer of the session's messages.
        """

    async def display_index_of(
        self, session_id: str, sequence_number: int
    ) -> int | None:
        """0-based rank of a persisted display row in the session's full history.

        This is the same rank ``get_history_page`` stamps as
        ``metadata.display_index``; clients key message ids and feedback on it.
        A store that cannot answer returns None, and the client falls back to
        its own approximation.
        """
        return None

    @abstractmethod
    async def find_reaction_session(
        self,
        *,
        user_id: str,
        session_key: str,
        not_active_before: datetime | None = None,
    ) -> Session | None:
        """The most recently active special session of `user_id` for one thread.

        `session_key` is the conversation id the Synapse dispatcher derived from
        the event envelope; every event of one conversation carries the same key,
        so the reaction handler can append to the existing thread instead of
        starting a new session. `not_active_before` drops threads idle for longer
        than the user's TTL. Returns None when there is no such thread — the
        caller then starts a fresh session.
        """

    @abstractmethod
    async def set_context_flag(
        self, session_id: str, sequence_numbers: list[int], is_context: bool
    ) -> int:
        """Flip ``is_context`` on already-persisted rows, whatever their is_display.

        ``save()`` only writes flags for rows in ``session.messages`` (display
        history). A summary is deliberately ``is_display=False``: after a
        reload it exists only in ``session.context``, so a demotion made by
        compression never reached the database and the stale summary came back
        on the next load. Returns the number of rows changed.
        """

    @abstractmethod
    async def archive_old_messages(self, session_id: str, keep_seq_threshold: int) -> int: ...

    @abstractmethod
    async def get_archived_messages(
        self, session_id: str, before_seq: int | None = None, limit: int = 50
    ) -> list[Message]: ...

    @abstractmethod
    async def get_or_create_share(self, session_id: str, owner_id: str | None) -> dict | None:
        """The session's public read-only share token, creating it on first call.

        One share per session: pressing "share" again must hand back the same
        link, since there is no list to go back to. ``None`` means the session
        does not exist (nothing to share).

        The row is ``{"token", "session_id", "owner_id", "created_at"}``.
        """

    @abstractmethod
    async def get_share(self, token: str) -> dict | None:
        """Resolve a share token to its row, or None when it was never issued.

        Deleting the session removes its share with it (FK cascade on the PG
        store; the in-memory store unlinks on delete).
        """


class InMemorySessionStore(SessionStore):
    def __init__(self) -> None:
        self._sessions: dict[str, Session] = {}
        self._shares: dict[str, dict] = {}  # session_id -> share row

    async def create(self, profile_id: str, user_id: str | None = None) -> Session:
        session = Session(profile_id=profile_id, user_id=user_id)
        self._sessions[session.id] = session
        return session

    async def get(self, session_id: str) -> Session | None:
        return self._sessions.get(session_id)

    async def save(self, session: Session) -> None:
        session.last_active = datetime.now(timezone.utc)
        self._sessions[session.id] = session

    @staticmethod
    def _special_filter(sessions: list["Session"], special: bool | None) -> list["Session"]:
        if special is None:
            return sessions
        return [s for s in sessions if s.special == special]

    async def list_all(
        self, user_id: str | None = None, is_admin: bool = False, special: bool | None = None
    ) -> list[Session]:
        sessions = self._sessions.values()
        if is_admin:
            pass
        elif user_id is not None:
            sessions = [s for s in sessions if s.user_id == user_id]
        else:
            # user_id=None non-admin means the anonymous local user: NULL-owner rows only.
            sessions = [s for s in sessions if s.user_id is None]
        return sorted(
            self._special_filter(list(sessions), special),
            key=lambda s: (s.pinned, s.last_active),
            reverse=True,
        )

    async def list_page(
        self,
        *,
        limit: int,
        offset: int,
        profile_id: str | None = None,
        user_id: str | None = None,
        is_admin: bool = False,
        special: bool | None = None,
    ) -> list[Session]:
        sessions = await self.list_all(user_id=user_id, is_admin=is_admin, special=special)
        if profile_id:
            sessions = [s for s in sessions if s.profile_id == profile_id]
        return sessions[offset:offset + limit]

    async def count_all(
        self,
        *,
        user_id: str | None = None,
        is_admin: bool = False,
        search: str | None = None,
        special: bool | None = None,
    ) -> int:
        sessions = await self.search_list(
            limit=10000, offset=0, user_id=user_id, is_admin=is_admin, search=search, special=special
        )
        return len(sessions)

    async def search_list(
        self,
        *,
        limit: int,
        offset: int,
        user_id: str | None = None,
        is_admin: bool = False,
        search: str | None = None,
        sort_by: str = "last_active",
        sort_order: str = "desc",
        special: bool | None = None,
    ) -> list[Session]:
        sessions = list(self._sessions.values())
        if is_admin:
            pass
        elif user_id is not None:
            sessions = [s for s in sessions if s.user_id == user_id]
        else:
            # user_id=None non-admin: anonymous local user — NULL-owner rows only.
            sessions = [s for s in sessions if s.user_id is None]
        sessions = self._special_filter(sessions, special)
        if search:
            q = search.lower()
            sessions = [
                s
                for s in sessions
                if (s.id and q in s.id.lower())
                or (s.name and q in s.name.lower())
                or (s.user_id and q in s.user_id.lower())
                or (s.profile_id and q in s.profile_id.lower())
                or any(q in (m.content or "").lower() for m in s.messages)
            ]
        key_map = {
            "created_at": lambda s: s.created_at,
            "last_active": lambda s: s.last_active,
            "name": lambda s: (s.name or "").lower(),
            "message_count": lambda s: len(s.messages),
        }
        key = key_map.get(sort_by, key_map["last_active"])
        reverse = sort_order == "desc"
        sessions = sorted(sessions, key=key, reverse=reverse)
        return sessions[offset:offset + limit]

    async def search_matches(
        self, session_ids: list[str], search: str
    ) -> dict[str, tuple[list[int], str]]:
        # Everything is already in memory here, so the scan is the whole job.
        return {
            sid: match_info_from_messages(self._sessions[sid].messages, search)
            for sid in session_ids
            if sid in self._sessions
        }

    async def delete(self, session_id: str) -> bool:
        if session_id in self._sessions:
            del self._sessions[session_id]
            self._shares.pop(session_id, None)
            return True
        return False

    async def set_pinned(self, session_id: str, pinned: bool) -> bool:
        s = self._sessions.get(session_id)
        if s is None:
            return False
        s.pinned = pinned
        return True

    async def set_name(self, session_id: str, name: str) -> bool:
        s = self._sessions.get(session_id)
        if s is None:
            return False
        s.name = name
        return True

    async def set_profile(self, session_id: str, profile_id: str) -> bool:
        s = self._sessions.get(session_id)
        if s is None:
            return False
        s.profile_id = profile_id
        return True

    async def find_reaction_session(
        self,
        *,
        user_id: str,
        session_key: str,
        not_active_before: datetime | None = None,
    ) -> Session | None:
        candidates = [
            s for s in self._sessions.values()
            if s.special
            and s.user_id == user_id
            and (s.session_metadata.get("synapse") or {}).get("session_key") == session_key
            and (not_active_before is None or s.last_active >= not_active_before)
        ]
        if not candidates:
            return None
        return max(candidates, key=lambda s: s.last_active)

    async def set_context_flag(
        self, session_id: str, sequence_numbers: list[int], is_context: bool
    ) -> int:
        session = self._sessions.get(session_id)
        if session is None:
            return 0
        wanted = set(sequence_numbers)
        changed = 0
        for msg in [*session.messages, *session.context]:
            if msg.sequence_number in wanted and msg.is_context != is_context:
                msg.is_context = is_context
                changed += 1
        return changed

    async def archive_old_messages(self, session_id: str, keep_seq_threshold: int) -> int:
        # In-memory store: no-op, everything stays in RAM
        return 0

    async def get_archived_messages(
        self, session_id: str, before_seq: int | None = None, limit: int = 50
    ) -> list[Message]:
        # In-memory store: no archive; return empty
        return []

    async def get_or_create_share(self, session_id: str, owner_id: str | None) -> dict | None:
        if session_id not in self._sessions:
            return None
        share = self._shares.get(session_id)
        if share is None:
            share = {
                "token": secrets.token_urlsafe(32),
                "session_id": session_id,
                "owner_id": owner_id,
                "created_at": datetime.now(timezone.utc),
            }
            self._shares[session_id] = share
        return dict(share)

    async def get_share(self, token: str) -> dict | None:
        for share in self._shares.values():
            if share["token"] == token:
                return dict(share)
        return None