# @File: backend/services/whatsapp/presence.py
# @Desc: In-memory hub for WhatsApp client chat realtime:
#        - per-subscription WebSocket fan-out
#        - typing indicators (with TTL so stale typings auto-clear)
#        - online/last-seen tracking for contacts
#        - new-message broadcast used by the chat service after a row is
#          inserted into `whatsapp_messages`
#
# Intentionally single-process only. A multi-worker deployment would swap
# this for Redis pub/sub, but for the MVP a single instance is sufficient
# and keeps the surface small.
from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Optional

from fastapi import WebSocket

logger = logging.getLogger(__name__)

# How long a typing event stays "alive" after the last keystroke. The client
# sends a refresh every few seconds while the user is typing; anything older
# than this is considered stale and swept.
TYPING_TTL_SECONDS = 6.0

# How long an inbound event counts the contact as "online" for. After this
# window elapses without any activity the contact flips back to "offline"
# with a last-seen timestamp.
ONLINE_TTL_SECONDS = 45.0

# Sweep cadence — small enough to feel responsive, big enough to stay cheap.
SWEEPER_INTERVAL = 2.0


@dataclass
class _TypingState:
    contact_id: int
    expires_at: float


@dataclass
class _PresenceState:
    contact_id: int
    online: bool = False
    last_seen_at: Optional[float] = None
    expires_at: float = 0.0


@dataclass
class _Channel:
    """Per-subscription fan-out state."""

    sockets: set[WebSocket] = field(default_factory=set)
    typing: dict[int, _TypingState] = field(default_factory=dict)
    presence: dict[int, _PresenceState] = field(default_factory=dict)


class PresenceHub:
    """In-process registry of active chat subscribers."""

    def __init__(self) -> None:
        self._channels: dict[int, _Channel] = {}
        self._lock = asyncio.Lock()
        self._sweeper_task: Optional[asyncio.Task[None]] = None

    # -- lifecycle -----------------------------------------------------

    def _ensure_sweeper(self) -> None:
        if self._sweeper_task is None or self._sweeper_task.done():
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                return
            self._sweeper_task = loop.create_task(self._sweep_forever())

    async def _sweep_forever(self) -> None:
        while True:
            try:
                await asyncio.sleep(SWEEPER_INTERVAL)
                await self._sweep_once()
            except asyncio.CancelledError:
                raise
            except Exception:  # noqa: BLE001
                logger.exception("presence sweeper error")

    async def _sweep_once(self) -> None:
        now = time.time()
        async with self._lock:
            channels = list(self._channels.items())

        for sub_id, channel in channels:
            expired_typing: list[int] = []
            for contact_id, state in list(channel.typing.items()):
                if state.expires_at <= now:
                    expired_typing.append(contact_id)

            flipped_offline: list[_PresenceState] = []
            for contact_id, state in list(channel.presence.items()):
                if state.online and state.expires_at <= now:
                    state.online = False
                    state.last_seen_at = state.last_seen_at or now
                    flipped_offline.append(state)

            for contact_id in expired_typing:
                channel.typing.pop(contact_id, None)
                await self._broadcast(
                    sub_id,
                    {
                        "type": "typing",
                        "contact_id": contact_id,
                        "typing": False,
                    },
                )

            for state in flipped_offline:
                await self._broadcast(
                    sub_id,
                    {
                        "type": "presence",
                        "contact_id": state.contact_id,
                        "online": False,
                        "last_seen_at": (
                            _iso(state.last_seen_at)
                            if state.last_seen_at is not None
                            else None
                        ),
                    },
                )

    # -- socket management --------------------------------------------

    async def register(self, subscription_id: int, websocket: WebSocket) -> None:
        async with self._lock:
            channel = self._channels.setdefault(subscription_id, _Channel())
            channel.sockets.add(websocket)
        self._ensure_sweeper()
        await self._send_snapshot(subscription_id, websocket)

    async def unregister(self, subscription_id: int, websocket: WebSocket) -> None:
        async with self._lock:
            channel = self._channels.get(subscription_id)
            if not channel:
                return
            channel.sockets.discard(websocket)
            if (
                not channel.sockets
                and not channel.typing
                and not channel.presence
            ):
                self._channels.pop(subscription_id, None)

    # -- snapshots ----------------------------------------------------

    async def _send_snapshot(
        self, subscription_id: int, websocket: WebSocket
    ) -> None:
        """Send current presence + typing state to a freshly-connected socket
        so the UI does not wait for the next event to paint."""
        now = time.time()
        async with self._lock:
            channel = self._channels.get(subscription_id)
            if not channel:
                return
            presence_snapshot = [
                {
                    "contact_id": state.contact_id,
                    "online": state.online and state.expires_at > now,
                    "last_seen_at": (
                        _iso(state.last_seen_at)
                        if state.last_seen_at is not None
                        else None
                    ),
                }
                for state in channel.presence.values()
            ]
            typing_snapshot = [
                state.contact_id
                for state in channel.typing.values()
                if state.expires_at > now
            ]
        try:
            await websocket.send_json(
                {
                    "type": "snapshot",
                    "presence": presence_snapshot,
                    "typing_contact_ids": typing_snapshot,
                    "server_time": _iso(now),
                }
            )
        except Exception:  # noqa: BLE001
            logger.debug("failed to send snapshot", exc_info=True)

    # -- event publishers ---------------------------------------------

    async def publish_typing(
        self,
        subscription_id: int,
        contact_id: int,
        *,
        typing: bool,
    ) -> None:
        now = time.time()
        async with self._lock:
            channel = self._channels.setdefault(subscription_id, _Channel())
            if typing:
                channel.typing[contact_id] = _TypingState(
                    contact_id=contact_id,
                    expires_at=now + TYPING_TTL_SECONDS,
                )
            else:
                channel.typing.pop(contact_id, None)

        await self._broadcast(
            subscription_id,
            {
                "type": "typing",
                "contact_id": contact_id,
                "typing": bool(typing),
            },
        )

    async def publish_presence(
        self,
        subscription_id: int,
        contact_id: int,
        *,
        online: bool,
        last_seen_at: Optional[float] = None,
    ) -> None:
        now = time.time()
        last_seen = last_seen_at if last_seen_at is not None else now
        async with self._lock:
            channel = self._channels.setdefault(subscription_id, _Channel())
            state = channel.presence.get(contact_id)
            if state is None:
                state = _PresenceState(contact_id=contact_id)
                channel.presence[contact_id] = state
            state.online = online
            state.last_seen_at = last_seen
            state.expires_at = now + ONLINE_TTL_SECONDS if online else now

        await self._broadcast(
            subscription_id,
            {
                "type": "presence",
                "contact_id": contact_id,
                "online": bool(online),
                "last_seen_at": _iso(last_seen),
            },
        )

    async def publish_message(
        self,
        subscription_id: int,
        message: dict,
    ) -> None:
        """Broadcast a freshly-inserted row from `whatsapp_messages`. Also
        refreshes the contact's presence as a side-effect when the message is
        inbound, since receiving a message implies the contact is online."""
        contact_id = message.get("contact_id")
        direction = message.get("direction")
        if contact_id and direction == "inbound":
            await self.publish_presence(
                subscription_id, int(contact_id), online=True
            )

        await self._broadcast(
            subscription_id,
            {
                "type": "message",
                "message": message,
            },
        )

    # -- low level ----------------------------------------------------

    async def _broadcast(self, subscription_id: int, payload: dict) -> None:
        async with self._lock:
            channel = self._channels.get(subscription_id)
            sockets = list(channel.sockets) if channel else []

        stale: list[WebSocket] = []
        for ws in sockets:
            try:
                await ws.send_json(payload)
            except Exception:  # noqa: BLE001
                stale.append(ws)

        if stale:
            async with self._lock:
                channel = self._channels.get(subscription_id)
                if channel:
                    for ws in stale:
                        channel.sockets.discard(ws)


def _iso(ts: Optional[float]) -> Optional[str]:
    if ts is None:
        return None
    import datetime as _dt

    return _dt.datetime.utcfromtimestamp(ts).isoformat() + "Z"


# Module-level singleton. Routers / services import this directly.
presence_hub = PresenceHub()


def get_presence_hub() -> PresenceHub:
    return presence_hub


async def publish_outbound_message_event(
    subscription_id: int, message: dict
) -> None:
    """Convenience wrapper used by the chat send path. Kept thin so callers
    don't import the hub module directly — easier to stub out in tests."""
    await presence_hub.publish_message(subscription_id, message)