straywild-discovery-server/netfishing_discovery/social_registry.py

"""Ephemeral, capability-addressed friend presence and live invitations."""

from __future__ import annotations

from dataclasses import dataclass
import hashlib
import secrets
import threading
import time
import unicodedata
from collections.abc import Callable, Mapping
from typing import Any

from .registry import RoomAdvertisement, RoomNotFoundError, RoomRegistry, ValidationError


PRESENCE_DOMAIN = "NETFISHING_PRESENCE_V1:"
INVITE_DOMAIN = "NETFISHING_INVITE_V1:"
SOCIAL_TOKEN_LENGTH = 64
MAX_SOCIAL_CHANNELS_PER_REQUEST = 200
MAX_SOCIAL_RECORDS = 8192
MAX_SOCIAL_RECORDS_PER_ADDRESS = 512
MAX_INVITES_PER_INBOX = 8
PRESENCE_TTL_SECONDS = 45.0
INBOX_REACHABLE_SECONDS = 15.0
INVITE_TTL_SECONDS = 30.0


class FriendOfflineError(Exception):
    pass


@dataclass(frozen=True, slots=True)
class Presence:
    channel: str
    display_name: str
    room_id: str
    game_version: str
    protocol_version: int
    address: str
    expires_at: float


@dataclass(frozen=True, slots=True)
class Invitation:
    invite_id: str
    inbox_id: str
    room_id: str
    expires_at: float


def _capability_id(domain: str, token: str) -> str:
    return hashlib.sha256(f"{domain}{token}".encode("utf-8")).hexdigest()


def _clean_token(value: Any, field_name: str) -> str:
    if not isinstance(value, str) or len(value) != SOCIAL_TOKEN_LENGTH:
        raise ValidationError(f"{field_name} must be a 64-character capability")
    if any(character not in "0123456789abcdef" for character in value):
        raise ValidationError(f"{field_name} must contain lowercase hexadecimal characters")
    return value


def _clean_token_list(value: Any, field_name: str) -> list[str]:
    if not isinstance(value, list) or len(value) > MAX_SOCIAL_CHANNELS_PER_REQUEST:
        raise ValidationError(
            f"{field_name} must contain at most {MAX_SOCIAL_CHANNELS_PER_REQUEST} entries"
        )
    result: list[str] = []
    for raw_token in value:
        token = _clean_token(raw_token, field_name)
        if token not in result:
            result.append(token)
    return result


def _clean_display_name(value: Any) -> str:
    if not isinstance(value, str):
        raise ValidationError("display_name must be a string")
    result = unicodedata.normalize("NFC", value).strip()
    if not result or len(result) > 24:
        raise ValidationError("display_name must contain between 1 and 24 characters")
    if any(unicodedata.category(character).startswith("C") for character in result):
        raise ValidationError("display_name contains unsupported control characters")
    return result


def _clean_protocol(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValidationError("protocol_version must be a positive integer")
    return value


class SocialRegistry:
    """Stores no accounts and no durable data; every record expires in seconds."""

    def __init__(self, clock: Callable[[], float] = time.time) -> None:
        self._clock = clock
        self._presence: dict[str, Presence] = {}
        self._inbox_polls: dict[str, float] = {}
        self._invites: dict[str, list[Invitation]] = {}
        self._lock = threading.RLock()

    def publish_presence(
        self, address: str, payload: Mapping[str, Any]
    ) -> list[str]:
        write_tokens = _clean_token_list(payload.get("write_tokens"), "write_tokens")
        online = payload.get("online")
        if not isinstance(online, bool):
            raise ValidationError("online must be a boolean")
        channels = [_capability_id(PRESENCE_DOMAIN, token) for token in write_tokens]
        now = self._clock()
        with self._lock:
            self._purge_locked(now)
            if not online:
                for channel in channels:
                    current = self._presence.get(channel)
                    if current is not None and current.address == address:
                        del self._presence[channel]
                return channels
            display_name = _clean_display_name(payload.get("display_name"))
            game_version = payload.get("game_version")
            if not isinstance(game_version, str) or not game_version.strip():
                raise ValidationError("game_version must be a non-empty string")
            protocol_version = _clean_protocol(payload.get("protocol_version"))
            room_id = payload.get("room_id", "")
            if not isinstance(room_id, str) or len(room_id) > 64 or "/" in room_id:
                raise ValidationError("room_id is invalid")
            new_channels = [channel for channel in channels if channel not in self._presence]
            address_count = sum(
                1 for record in self._presence.values() if record.address == address
            )
            if len(self._presence) + len(new_channels) > MAX_SOCIAL_RECORDS:
                raise ValidationError("the live presence directory is at capacity")
            if address_count + len(new_channels) > MAX_SOCIAL_RECORDS_PER_ADDRESS:
                raise ValidationError("too many live presence channels from this address")
            for channel in channels:
                self._presence[channel] = Presence(
                    channel=channel,
                    display_name=display_name,
                    room_id=room_id,
                    game_version=game_version.strip(),
                    protocol_version=protocol_version,
                    address=address,
                    expires_at=now + PRESENCE_TTL_SECONDS,
                )
        return channels

    def query_presence(
        self,
        channels_value: Any,
        game_version: str,
        protocol_version: int,
        rooms: RoomRegistry,
    ) -> list[dict[str, Any]]:
        channels = _clean_token_list(channels_value, "channels")
        now = self._clock()
        result: list[dict[str, Any]] = []
        with self._lock:
            self._purge_locked(now)
            records = [self._presence.get(channel) for channel in channels]
        for record in records:
            if (
                record is None
                or record.game_version != game_version
                or record.protocol_version != protocol_version
            ):
                continue
            room = rooms.get_public_room(
                record.room_id,
                game_version=game_version,
                protocol_version=protocol_version,
            ) if record.room_id else None
            result.append(
                {
                    "channel": record.channel,
                    "display_name": record.display_name,
                    "online": True,
                    "room": room.public_dict() if room is not None else {},
                }
            )
        return result

    def poll_invitations(
        self,
        inbox_tokens_value: Any,
        game_version: str,
        protocol_version: int,
        rooms: RoomRegistry,
    ) -> list[dict[str, Any]]:
        tokens = _clean_token_list(inbox_tokens_value, "inbox_tokens")
        inbox_ids = [_capability_id(INVITE_DOMAIN, token) for token in tokens]
        now = self._clock()
        invitations: list[Invitation] = []
        with self._lock:
            self._purge_locked(now)
            for inbox_id in inbox_ids:
                self._inbox_polls[inbox_id] = now
                invitations.extend(self._invites.pop(inbox_id, []))
        result: list[dict[str, Any]] = []
        for invitation in invitations:
            room = rooms.get_public_room(
                invitation.room_id,
                game_version=game_version,
                protocol_version=protocol_version,
            )
            if room is None:
                continue
            result.append(
                {
                    "invite_id": invitation.invite_id,
                    "inbox_id": invitation.inbox_id,
                    "room": room.public_dict(),
                }
            )
        return result

    def send_invitation(
        self,
        inbox_token_value: Any,
        room_id: str,
        game_version: str,
        protocol_version: int,
        rooms: RoomRegistry,
    ) -> Invitation:
        inbox_token = _clean_token(inbox_token_value, "inbox_token")
        if not isinstance(room_id, str) or not room_id or len(room_id) > 64:
            raise ValidationError("room_id is invalid")
        room = rooms.get_public_room(
            room_id,
            game_version=game_version,
            protocol_version=protocol_version,
        )
        if room is None:
            raise RoomNotFoundError("room is not available")
        inbox_id = _capability_id(INVITE_DOMAIN, inbox_token)
        now = self._clock()
        with self._lock:
            self._purge_locked(now)
            last_poll = self._inbox_polls.get(inbox_id, 0.0)
            if now - last_poll > INBOX_REACHABLE_SECONDS:
                raise FriendOfflineError("This person needs to be online to do this.")
            queue = self._invites.setdefault(inbox_id, [])
            if len(queue) >= MAX_INVITES_PER_INBOX:
                queue.pop(0)
            invitation = Invitation(
                invite_id=secrets.token_hex(16),
                inbox_id=inbox_id,
                room_id=room.room_id,
                expires_at=now + INVITE_TTL_SECONDS,
            )
            queue.append(invitation)
            return invitation

    def counts(self) -> tuple[int, int]:
        now = self._clock()
        with self._lock:
            self._purge_locked(now)
            return len(self._presence), sum(len(queue) for queue in self._invites.values())

    def _purge_locked(self, now: float) -> None:
        self._presence = {
            channel: record
            for channel, record in self._presence.items()
            if record.expires_at > now
        }
        self._inbox_polls = {
            inbox_id: polled_at
            for inbox_id, polled_at in self._inbox_polls.items()
            if now - polled_at <= INBOX_REACHABLE_SECONDS
        }
        for inbox_id in list(self._invites):
            queue = [invite for invite in self._invites[inbox_id] if invite.expires_at > now]
            if queue:
                self._invites[inbox_id] = queue
            else:
                del self._invites[inbox_id]