"""Authority-local subscription ordering over the existing bounded transports/replay."""
from copy import deepcopy
import logging
import threading
import uuid

from tui_gateway import event_replay
from tui_gateway.transport import FanoutTransport


class SessionEvents:
    def __init__(self):
        self.lock = threading.RLock()
        self.fanout = FanoutTransport()
        # A private lifetime key prevents two profiles (or legacy TUI owners)
        # with the same persisted ID from reading each other's replay.
        self._key = uuid.uuid4().hex
        self.epoch = uuid.uuid4().hex
        self.sequence = 0
        self._public_id = None
        self.execution = {}
        # Synchronous same-thread recipients (API run projections): unlike fanout peers they
        # cannot lose a frame to a detach that races the writer thread.
        self.observers = set()
        # Set by the authority that attaches transports: retires the subscription of a
        # peer whose bounded backlog overflowed, so it cannot linger looking subscribed.
        self.on_overflow = None

    def watermark(self):
        with self.lock:
            if self.sequence and not event_replay.latest_seq(self._key):
                # The shared bounded cache evicted this session. Never reuse a
                # sequence in the old epoch, even if publication resumes later.
                self.epoch = uuid.uuid4().hex
                self.sequence = 0
            return self.epoch, self.sequence

    def publish(self, session_id, payload, *, event_type="message.complete", execution=None):
        # ``execution`` stamps this frame alone; the shared stamp is read without the lock elsewhere.
        with self.lock:
            self.watermark()
            frame = {'jsonrpc': '2.0', 'method': 'event', 'params': {
                **(self.execution if execution is None else execution), 'type': event_type, 'session_id': self._key,
                'payload': deepcopy(payload), 'replay_epoch': self.epoch}}
            event_replay._stamp_event(frame)
            if frame['params']['seq'] <= self.sequence:
                # Another publisher may evict us after watermark's lookup.
                self.epoch = uuid.uuid4().hex
                frame['params']['replay_epoch'] = self.epoch
            # The ring froze the frame under its private lookup key; live recipients see the
            # canonical ID here and ``since`` restores it on the thawed replay copies.
            frame['params']['session_id'] = session_id
            self._public_id = session_id
            self.sequence = frame['params']['seq']

            def overflow(transport):
                # The peer's socket is healthy; only its event subscription ended. Same
                # shape as the client's own synthetic gap so one consumer handles both.
                if self.on_overflow is not None:
                    self.on_overflow(transport)
                return {'jsonrpc': '2.0', 'method': 'event', 'params': {
                    'type': 'session.replay_gap', 'session_id': session_id,
                    'payload': {'replay_epoch': self.epoch, 'latest_seq': self.sequence}}}
            delivered = self.fanout.write(frame, overflow=overflow)
            for observer in tuple(self.observers):
                try:
                    observer(frame)
                    delivered = True
                except Exception:
                    # A request-owned projection cannot undo committed publication or starve peers.
                    logging.getLogger(__name__).warning('Session event observer failed for %s', session_id,
                                                        exc_info=True)
            # True when a live viewer or observer took the frame; the replay ring alone is not
            # delivery (a detached finite viewer reads nothing until it reattaches).
            return delivered

    def since(self, epoch, sequence):
        with self.lock:
            current_epoch, latest = self.watermark()
            frames = event_replay.events_since(self._key, sequence)
            missing = (epoch != current_epoch or sequence > latest
                       or event_replay.is_truncated(self._key, sequence)
                       or event_replay.latest_seq(self._key) != latest)
            events = [] if missing else [
                {**frame, 'session_id': self._public_id} if frame.get('session_id') == self._key else frame
                for frame in frames]
            return {'events': events, 'latest_seq': latest, 'last_sequence': latest,
                    'epoch': current_epoch, 'replay_epoch': current_epoch,
                    'truncated': missing, 'snapshot_required': missing, 'count': len(events)}