PeerSimulation.cs

using Decentraland.Common;
using Decentraland.Pulse;
using Pulse.InterestManagement;
using Pulse.Messaging;
using Pulse.Transport;
using static Pulse.Messaging.MessagePipe;

namespace Pulse.Peers.Simulation;

/// <summary>
///     Per-worker simulation step. Iterates authenticated observers owned by this worker,
///     queries <see cref="IAreaOfInterest" /> for visible subjects, diffs snapshots,
///     and sends STATE_DELTA / STATE_FULL via <see cref="MessagePipe" />.
///     Instantiated once per worker - not shared across workers: thus thread-safety is ensured without concurrency.
/// </summary>
public sealed class PeerSimulation : IPeerSimulation
{
    public const string SELF_MIRROR_WALLET_ID = "self_mirror";
    /// <summary>
    ///     Sweep stale views every N ticks to reclaim memory from subjects that left the interest set.
    /// </summary>
    private const uint SWEEP_INTERVAL = 100;

    /// <summary>
    ///     Per-observer views: observer PeerIndex → (subject PeerIndex → view).
    ///     Stored here, exclusive to this worker — no locks.
    /// </summary>
    internal readonly Dictionary<PeerIndex, Dictionary<PeerIndex, PeerToPeerView>> observerViews = new ();

    private readonly IAreaOfInterest areaOfInterest;
    private readonly SnapshotBoard snapshotBoard;
    private readonly SpatialGrid spatialGrid;
    private readonly IdentityBoard identityBoard;
    private readonly MessagePipe messagePipe;
    private readonly ITimeProvider timeProvider;
    private readonly ITransport transport;
    private readonly ProfileBoard profileBoard;
    private readonly IPeerIndexAllocator peerIndexAllocator;
    private readonly ILogger<PeerSimulation> logger;
    private readonly bool selfMirrorEnabled;
    private readonly PeerViewSimulationTier selfMirrorTier;
    private readonly bool resyncWithDelta;
    private readonly uint disconnectionCleanTimeoutMs;
    private readonly uint pendingAuthCleanTimeoutMs;

    /// <summary>
    ///     Reusable collector to avoid allocation per tick.
    /// </summary>
    private readonly InterestCollector collector = new ();

    /// <summary>
    ///     Pre-computed divisors: SimulationSteps[tier] / baseTickMs.
    ///     TIER_0 → 1 (every tick), TIER_1 → 2 (every 2nd tick), TIER_2 → 4 (every 4th tick).
    /// </summary>
    private readonly uint[] tierDivisors;

    /// <summary>
    ///     Reusable buffer for sweep removals — avoids allocating a list every sweep.
    /// </summary>
    private readonly List<PeerIndex> sweepBuffer = new ();

    private readonly HashSet<PeerIndex> peersToBeRemoved = new ();

    public uint BaseTickMs { get; }

    public PeerSimulation(
        IAreaOfInterest areaOfInterest,
        SnapshotBoard snapshotBoard,
        SpatialGrid spatialGrid,
        IdentityBoard identityBoard,
        MessagePipe messagePipe,
        uint[] simulationSteps,
        ITimeProvider timeProvider,
        ITransport transport,
        ProfileBoard profileBoard,
        IPeerIndexAllocator peerIndexAllocator,
        ILogger<PeerSimulation> logger,
        bool selfMirrorEnabled = false,
        int selfMirrorTier = 0,
        bool resyncWithDelta = false,
        uint disconnectionCleanTimeoutMs = 5000,
        uint pendingAuthCleanTimeoutMs = 30000)
    {
        this.areaOfInterest = areaOfInterest;
        this.snapshotBoard = snapshotBoard;
        this.spatialGrid = spatialGrid;
        this.identityBoard = identityBoard;
        this.messagePipe = messagePipe;
        this.timeProvider = timeProvider;
        this.transport = transport;
        this.profileBoard = profileBoard;
        this.peerIndexAllocator = peerIndexAllocator;
        this.logger = logger;
        this.selfMirrorEnabled = selfMirrorEnabled;
        this.selfMirrorTier = new PeerViewSimulationTier((byte)selfMirrorTier);
        this.resyncWithDelta = resyncWithDelta;
        this.disconnectionCleanTimeoutMs = disconnectionCleanTimeoutMs;
        this.pendingAuthCleanTimeoutMs = pendingAuthCleanTimeoutMs;

        BaseTickMs = simulationSteps[0];
        tierDivisors = new uint[simulationSteps.Length];

        for (var i = 0; i < simulationSteps.Length; i++)
            tierDivisors[i] = simulationSteps[i] / BaseTickMs;
    }

    /// <summary>
    ///     Runs one simulation tick for all authenticated observers in the given peer set.
    /// </summary>
    public void SimulateTick(Dictionary<PeerIndex, PeerState> peers, uint tickCounter)
    {
        foreach ((PeerIndex observerId, PeerState observerState) in peers)
        {
            if (observerState.ConnectionState == PeerConnectionState.PENDING_AUTH)
            {
                if (timeProvider.MonotonicTime - observerState.TransportState.ConnectionTime >= pendingAuthCleanTimeoutMs)
                {
                    transport.Disconnect(observerId, DisconnectReason.AUTH_TIMEOUT);
                    logger.LogInformation("Peer {Peer} disconnected due to authentication timed out", observerId);
                    continue;
                }
            }

            if (observerState.ConnectionState == PeerConnectionState.DISCONNECTING)
            {
                if (timeProvider.MonotonicTime - observerState.TransportState.DisconnectionTime >= disconnectionCleanTimeoutMs)
                {
                    CleanupDisconnectedPeer(observerId);
                    continue;
                }
            }

            if (observerState.ConnectionState != PeerConnectionState.AUTHENTICATED)
                continue;

            if (!snapshotBoard.TryRead(observerId, out PeerSnapshot observerSnapshot))
                continue;

            if (!observerViews.TryGetValue(observerId, out Dictionary<PeerIndex, PeerToPeerView>? views))
            {
                views = new Dictionary<PeerIndex, PeerToPeerView>();
                observerViews[observerId] = views;
            }

            collector.Clear();
            areaOfInterest.GetVisibleSubjects(observerId, in observerSnapshot, collector);
            AddSelfMirror(observerId, in observerSnapshot);

            string? observerWallet = identityBoard.GetWalletIdByPeerIndex(observerId);

            ProcessVisibleSubjects(observerId, observerWallet, views, observerState.ResyncRequests, tickCounter);

            observerState.ResyncRequests?.Clear();

            if (tickCounter % SWEEP_INTERVAL == 0)
                SweepStaleViews(observerId, views, tickCounter);
        }

        foreach (PeerIndex pi in peersToBeRemoved)
            peers.Remove(pi);

        peersToBeRemoved.Clear();
    }

    /// <summary>
    ///     Call when a peer disconnects to clean up its observer views.
    /// </summary>
    public void RemoveObserver(PeerIndex observerId)
    {
        observerViews.Remove(observerId);
    }

    /// <summary>
    ///     Inject the observer as its own subject so it receives its own state under
    ///     <see cref="SELF_MIRROR_WALLET_ID" /> (a client-side animation-testing aid).
    ///     <para />
    ///     This bypasses the AoI, so it must re-apply the AoI's realm invariant itself: a peer
    ///     with no realm yet (legacy connect, before its first teleport) is invisible to every
    ///     observer — itself included — so it is not mirrored until a realm is set. Without this
    ///     guard the self-mirror would emit a <c>PlayerJoined</c> carrying a realm-less snapshot.
    /// </summary>
    private void AddSelfMirror(PeerIndex observerId, in PeerSnapshot observerSnapshot)
    {
        if (!selfMirrorEnabled || observerSnapshot.Realm == null)
            return;

        collector.Add(observerId, selfMirrorTier);
    }

    // ── Per-subject orchestration ───────────────────────────────────

    private void ProcessVisibleSubjects(
        PeerIndex observerId,
        string? observerWallet,
        Dictionary<PeerIndex, PeerToPeerView> views,
        Dictionary<PeerIndex, uint>? resyncRequests,
        uint tickCounter)
    {
        for (var i = 0; i < collector.Count; i++)
        {
            InterestEntry entry = collector.Entries[i];

            bool isSelfMirror = entry.Subject == observerId;

            if (isSelfMirror && !selfMirrorEnabled)
                continue;

            // Same-wallet ghost suppression: during the disconnect-cleanup window a stale
            // PeerIndex from the prior session still occupies SnapshotBoard/SpatialGrid/
            // IdentityBoard and surfaces to its own freshly reconnected PeerIndex as a
            // visible subject. The PeerIndex differs (allocator pending-recycle), so the
            // == observerId check above doesn't catch it. Skipping by wallet here prevents
            // the reconnecting client from receiving a PlayerJoined for itself.
            if (!isSelfMirror
                && observerWallet != null
                && string.Equals(
                    identityBoard.GetWalletIdByPeerIndex(entry.Subject),
                    observerWallet,
                    StringComparison.OrdinalIgnoreCase))
                continue;

            bool isNew = !views.TryGetValue(entry.Subject, out PeerToPeerView view);

            if (!isNew && DetectAndHandleAliasing(observerId, entry.Subject, isSelfMirror, view, views))
                isNew = true;

            // Stamp before tier gate — a TIER_2 subject fires every 4th tick,
            // but it's still visible on the intervening ticks. Without this,
            // 3 unstamped ticks would trigger false re-entry detection.
            if (!isNew)
            {
                view.LastSeenTick = tickCounter;
                views[entry.Subject] = view;
            }

            // Skip if this tier is not due on this tick — but never gate a pending resync
            bool hasResync = !isNew && resyncRequests != null && resyncRequests.ContainsKey(entry.Subject);
            int tierIndex = entry.Tier.Value;

            if (!hasResync && tierIndex < tierDivisors.Length && tickCounter % tierDivisors[tierIndex] != 0)
                continue;

            if (!snapshotBoard.TryRead(entry.Subject, out PeerSnapshot latestSnapshot))
                continue;

            if (isNew)
            {
                view = HandleNewSubject(observerId, entry.Subject, latestSnapshot, isSelfMirror, resyncRequests);
                view.LastSeenTick = tickCounter;
                views[entry.Subject] = view;
                continue;
            }

            TryAnnounceProfile(observerId, entry.Subject, ref view);

            PeerSnapshot lastSentState = ProcessExistingSubject(
                observerId, entry, ref view, latestSnapshot, resyncRequests);

            view.LastSentSnapshot = lastSentState;
            view.LastSeenTick = tickCounter;
            views[entry.Subject] = view;
        }
    }

    /// <summary>
    ///     Defense-in-depth against <see cref="PeerIndex" /> aliasing. If the observer's view
    ///     was seeded for a different wallet than the one currently occupying this slot, the
    ///     logical identity has been replaced mid-session — emit <c>PlayerLeft</c> for the
    ///     stale identity and drop the view so the caller re-enters the <c>isNew</c> path.
    ///     <para />
    ///     The transport-level <see cref="PeerIndexAllocator" /> prevents this via pending-
    ///     recycle, but the simulation must not rely on that invariant silently. Returns true
    ///     when aliasing was detected and the view was removed.
    /// </summary>
    private bool DetectAndHandleAliasing(
        PeerIndex observerId, PeerIndex subjectId, bool isSelfMirror,
        PeerToPeerView view, Dictionary<PeerIndex, PeerToPeerView> views)
    {
        string? currentWallet = isSelfMirror
            ? SELF_MIRROR_WALLET_ID
            : identityBoard.GetWalletIdByPeerIndex(subjectId);

        if (string.Equals(view.LastSentWalletId, currentWallet, StringComparison.OrdinalIgnoreCase))
            return false;

        messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
        {
            PlayerLeft = new PlayerLeft { SubjectId = subjectId },
        }, PacketMode.RELIABLE));

        logger.LogWarning(
            "PeerIndex {Subject} aliased (view held '{OldWallet}', board now '{NewWallet}') — observer {Observer} notified",
            subjectId, view.LastSentWalletId, currentWallet, observerId);

        views.Remove(subjectId);
        return true;
    }

    /// <summary>
    ///     First-time visibility: send PlayerJoined with full state.
    ///     If the subject is mid-emote (the ledger guarantees <see cref="PeerSnapshot.Emote" />
    ///     reflects current state), announce the emote immediately so the observer animates it
    ///     instead of waiting for the next start/stop event — which may never come for the
    ///     remainder of this emote.
    /// </summary>
    private PeerToPeerView HandleNewSubject(
        PeerIndex observerId, PeerIndex subjectId,
        PeerSnapshot latestSnapshot, bool isSelfMirror,
        Dictionary<PeerIndex, uint>? resyncRequests)
    {
        resyncRequests?.Remove(subjectId);

        int profileVersion = profileBoard.Get(subjectId);

        string? userId = isSelfMirror
            ? SELF_MIRROR_WALLET_ID
            : identityBoard.GetWalletIdByPeerIndex(subjectId);

        messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
        {
            PlayerJoined = new PlayerJoined
            {
                UserId = userId,
                ProfileVersion = profileVersion,
                State = CreateFullState(subjectId, latestSnapshot),
                Realm = latestSnapshot.Realm ?? string.Empty,
            },
        }, PacketMode.RELIABLE));

        logger.LogInformation("Sending PlayerJoined for subject {Subject} to observer {Observer}", subjectId, observerId);

        var view = new PeerToPeerView
        {
            Onto = subjectId,
            LastSentProfileVersion = profileVersion,
            LastSentTeleportSeq = latestSnapshot.Seq,
            LastSentSnapshot = latestSnapshot,
            LastSentWalletId = userId,

            // LastSentSeq is assigned below — either implicitly by SendEmoteStarted or explicitly.
        };

        // If the subject is already emoting when first visible, broadcast the ongoing emote
        // so the observer can scrub the animation forward instead of staying idle. Treated as
        // the eviction case: we only know the emote through the ledger-carried latest snapshot,
        // not a real EmoteStart event, so the tripwire should warn (not error) on seq collisions.
        if (latestSnapshot.Emote is { EmoteId: not null } activeEmote)
        {
            SendEmoteStarted(observerId, ref view, subjectId, latestSnapshot, activeEmote, fromEviction: true);
            view.LastSentEmote = activeEmote;
        }
        else
        {
            // No seq-carrying send happened — seed the tripwire baseline so the next SendTracked
            // can detect genuine duplicates against a meaningful prior seq.
            view.LastSentSeq = latestSnapshot.Seq;
        }

        return view;
    }

    /// <summary>
    ///     Processes an already-known subject: scans intermediates for discrete events,
    ///     syncs emote stop, then falls back to resync or delta.
    ///     Returns the snapshot that should become the new baseline.
    /// </summary>
    private PeerSnapshot ProcessExistingSubject(
        PeerIndex observerId,
        InterestEntry entry,
        ref PeerToPeerView view,
        PeerSnapshot latestSnapshot,
        Dictionary<PeerIndex, uint>? resyncRequests)
    {
        PeerSnapshot lastSentState = view.LastSentSnapshot;
        var discreteEventSent = false;

        // --- Phase 1: scan intermediates, collect last of each discrete event type ---
        ScanIntermediateEvents(entry.Subject, view.LastSentSnapshot.Seq, latestSnapshot,
            out PeerSnapshot? lastEmoteStart, out PeerSnapshot? lastEmoteStop, out PeerSnapshot? lastTeleport,
            out bool emoteStartFromEviction);

        // --- Broadcast teleport (spatial snap first) ---
        if (lastTeleport is { } tp && view.LastSentTeleportSeq < tp.Seq)
        {
            SendTeleport(observerId, ref view, entry.Subject, tp);
            resyncRequests?.Remove(entry.Subject);
            view.LastSentTeleportSeq = tp.Seq;
            lastSentState = tp;
            discreteEventSent = true;
        }

        // --- Broadcast emote start only if the emote is still active (not stopped in the same batch).
        //     An emote that started and stopped between ticks is invisible to the observer. ---
        bool emoteStartIsEffective = lastEmoteStart.HasValue
                                     && lastEmoteStart.Value.Seq > (lastEmoteStop?.Seq ?? 0);

        if (emoteStartIsEffective
            && lastEmoteStart!.Value.Emote is { EmoteId: not null } emote
            && !(emote.EmoteId == view.LastSentEmote?.EmoteId && emote.StartSeq == view.LastSentEmote?.StartSeq))
        {
            PeerSnapshot es = lastEmoteStart.Value;
            SendEmoteStarted(observerId, ref view, entry.Subject, es, emote, fromEviction: emoteStartFromEviction);
            resyncRequests?.Remove(entry.Subject);
            view.LastSentEmote = emote;

            if (es.Seq > lastSentState.Seq)
                lastSentState = es;

            discreteEventSent = true;
        }

        // --- Phase 2: sync emote stop (skip when the start is still effective —
        //     either just sent, or already synced via dedup — the emote is active) ---
        if (!emoteStartIsEffective)
            TrySyncEmoteStop(observerId, entry.Subject, ref view, ref lastSentState, lastEmoteStop);

        // --- Phase 3: resync or delta (skip if discrete events already carried full state) ---
        if (!discreteEventSent)
        {
            lastSentState = HandleResyncOrDelta(
                observerId, entry, ref view, lastSentState, latestSnapshot, resyncRequests);
        }

        return lastSentState;
    }

    /// <summary>
    ///     Collect the last teleport, emote start, and emote stop within
    ///     <paramref name="fromSeq" />+1..<paramref name="latestSnapshot" />.Seq.
    ///     <para />
    ///     Under the emote ledger (<see cref="SnapshotBoard.Publish" />), every snapshot between
    ///     EmoteStart and EmoteStop carries a non-null <c>Emote.EmoteId</c>. Most of those are
    ///     *carry-forwards* of an earlier start, not real start events. A real EmoteStart
    ///     snapshot is the only one where <c>Seq == Emote.StartSeq</c> — the handler stamps
    ///     both from the same fresh sequence number. Carry-forwards keep the original
    ///     <c>StartSeq</c> while their own <c>Seq</c> advances. Using Seq (not ServerTick) as
    ///     the discriminator is required because multiple snapshots can share a ServerTick when
    ///     e.g. a teleport and an emote-start are processed on the same tick; Seq is monotonic
    ///     and unique per snapshot. Detecting the real start via that equality lets
    ///     <see cref="SendEmoteStarted" /> broadcast the position the subject had *at the moment
    ///     they started emoting*, not a later carry-forward position.
    ///     <para />
    ///     Ring-wrap fallback: if the real start snapshot has been evicted from the ring (high
    ///     publish rate, low-tier observer), the scan can't find a <c>Seq == StartSeq</c>
    ///     match. In that case we promote the <b>earliest</b> carrying snapshot in range (the
    ///     one closest in time to the real start, therefore closest in position) to
    ///     <paramref name="lastEmoteStart" /> and set <paramref name="emoteStartFromEviction" />
    ///     so the caller can tag the subsequent <see cref="SendEmoteStarted" /> as an
    ///     expected-duplicate case. If nothing in range carries the emote but
    ///     <paramref name="latestSnapshot" /> does (entire scan range predated the start), we
    ///     fall back to it as a last resort.
    ///     <para />
    ///     Only the stop snapshot itself has <c>StopReason != null</c> (post-stop snapshots
    ///     inherit <c>null</c>), so <paramref name="lastEmoteStop" /> remains unique per transition.
    /// </summary>
    private void ScanIntermediateEvents(PeerIndex subjectId, uint fromSeq, PeerSnapshot latestSnapshot,
        out PeerSnapshot? lastEmoteStart, out PeerSnapshot? lastEmoteStop, out PeerSnapshot? lastTeleport,
        out bool emoteStartFromEviction)
    {
        lastEmoteStart = null;
        lastEmoteStop = null;
        lastTeleport = null;
        emoteStartFromEviction = false;

        PeerSnapshot? earliestCarry = null;

        for (uint seq = fromSeq + 1; seq <= latestSnapshot.Seq; seq++)
        {
            if (!snapshotBoard.TryRead(subjectId, seq, out PeerSnapshot snapshot))
                continue;

            if (snapshot.Emote is { EmoteId: not null, StartSeq: var startSeq })
            {
                if (snapshot.Seq == startSeq)
                {
                    lastEmoteStart = snapshot;

                    // Reset the earliest-carry tracker: any carries preceding a newer real start
                    // belonged to a superseded emote (the real start is what the observer must
                    // see; older carries are irrelevant).
                    earliestCarry = null;
                }
                else if (earliestCarry is null)
                {
                    // First carry we've seen since the last real start (or scan start) — hold
                    // onto it as the best fallback position if no real start appears.
                    earliestCarry = snapshot;
                }
            }

            if (snapshot.Emote is { StopReason: not null })
                lastEmoteStop = snapshot;

            if (snapshot.IsTeleport)
                lastTeleport = snapshot;
        }

        if (lastEmoteStart is null)
        {
            if (earliestCarry is { } carry)
            {
                lastEmoteStart = carry;
                emoteStartFromEviction = true;
            }
            else if (latestSnapshot.Emote is { EmoteId: not null, StopReason: null })
            {
                // Scan range was empty (e.g. single-tick gap with no new publishes) but the
                // latest snapshot outside the scan still holds the emote state.
                lastEmoteStart = latestSnapshot;
                emoteStartFromEviction = true;
            }
        }
    }

    // ── Emote stop detection ────────────────────────────────────────

    private void TrySyncEmoteStop(
        PeerIndex observerId, PeerIndex subjectId,
        ref PeerToPeerView view,
        ref PeerSnapshot lastSentState,
        PeerSnapshot? stopSnapshot)
    {
        if (view.LastSentEmote?.EmoteId == null)
            return;
        // Explicit stop — either Cancelled (from EmoteStopHandler) or Completed (from EmoteCompleter).
        // Both are published as real stop snapshots on the subject's worker, so they carry their own seq.
        if (stopSnapshot?.Emote is { StopReason: not null } stopEmote)
        {
            SendEmoteStopped(observerId, ref view, subjectId, stopSnapshot.Value, stopEmote.StopReason!.Value);
            view.LastSentEmote = null;

            // Advance the Phase 3 baseline to the stop snapshot — otherwise Phase 3's
            // SendDelta would diff from the pre-emote baseline and potentially re-send
            // the same seq already carried by EmoteStopped above.
            if (stopSnapshot.Value.Seq > lastSentState.Seq)
                lastSentState = stopSnapshot.Value;
        }
    }

    // ── Resync / delta ──────────────────────────────────────────────

    private PeerSnapshot HandleResyncOrDelta(
        PeerIndex observerId,
        InterestEntry entry,
        ref PeerToPeerView view,
        PeerSnapshot lastSentState,
        PeerSnapshot latestSnapshot,
        Dictionary<PeerIndex, uint>? resyncRequests)
    {
        if (resyncRequests == null || !resyncRequests.Remove(entry.Subject, out uint lastKnownSeq))
        {
            SendDelta(observerId, ref view, entry.Subject, lastSentState, latestSnapshot, entry.Tier, PacketMode.UNRELIABLE_SEQUENCED);
            return latestSnapshot;
        }

        // Try a targeted delta from the client's baseline; fall back to full state
        // if the baseline is evicted, the seq hasn't advanced, or the feature is disabled.
        if (resyncWithDelta
            && snapshotBoard.TryRead(entry.Subject, lastKnownSeq, out PeerSnapshot knownSnapshot)
            && knownSnapshot.Seq != latestSnapshot.Seq)
        {
            SendDelta(observerId, ref view, entry.Subject, knownSnapshot, latestSnapshot, entry.Tier, PacketMode.RELIABLE, fromResync: true);

            logger.LogInformation("Resync fulfilled with targeted delta for subject {Subject} to observer {Observer} (lastKnownSeq={LastKnownSeq})",
                entry.Subject, observerId, lastKnownSeq);
        }
        else
        {
            SendTracked(observerId, ref view, latestSnapshot.Seq, new ServerMessage
            {
                PlayerStateFull = CreateFullState(entry.Subject, latestSnapshot),
            }, PacketMode.RELIABLE, fromResync: true);

            logger.LogWarning("Resync fallback to STATE_FULL for subject {Subject} to observer {Observer} (lastKnownSeq={LastKnownSeq}, gap={SeqGap})",
                entry.Subject, observerId, lastKnownSeq, latestSnapshot.Seq - lastKnownSeq);
        }

        return latestSnapshot;
    }

    // ── Message sending ─────────────────────────────────────────────

    /// <summary>
    ///     Sends a seq-carrying message to the observer and records the seq on the view as a
    ///     duplicate-delivery tripwire. Logs an error if <paramref name="seq" /> equals
    ///     <see cref="PeerToPeerView.LastSentSeq" /> — that means the same sequence number
    ///     has already been delivered to this observer for this subject, which indicates a
    ///     bug in the simulation pipeline.
    ///     <para />
    ///     When <paramref name="fromEmoteStartEviction" /> is <c>true</c>, the duplicate is
    ///     expected: the EmoteStart snapshot was evicted from the ring, the scan fell back to
    ///     the latest carrying snapshot, and its seq may collide with a prior send in the same
    ///     tick (e.g. a teleport also landing at the latest seq). The collision is logged as
    ///     a warning with explicit eviction context rather than an error.
    ///     <para />
    ///     When <paramref name="fromResync" /> is <c>true</c>, the duplicate is also expected:
    ///     the resync path retransmits the latest known seq over the reliable channel to fill
    ///     a client-side gap, and that seq may match a prior unreliable send that the client
    ///     missed. The caller already logs the resync context as a warning, so the tripwire
    ///     stays silent in this case.
    /// </summary>
    private void SendTracked(PeerIndex observerId, ref PeerToPeerView view, uint seq, ServerMessage message, PacketMode packetMode,
        bool fromEmoteStartEviction = false,
        bool fromResync = false)
    {
        if (seq == view.LastSentSeq && !fromResync)
        {
            if (fromEmoteStartEviction)
                logger.LogWarning(
                    "Duplicate seq {Seq} sent to observer {Observer} for subject {Subject} ({MessageCase}) — EmoteStart snapshot evicted from ring, scan fell back to the latest carrying snapshot",
                    seq, observerId, view.Onto, message.MessageCase);
            else
                logger.LogError(
                    "Duplicate seq {Seq} sent to observer {Observer} for subject {Subject} ({MessageCase})",
                    seq, observerId, view.Onto, message.MessageCase);
        }

        view.LastSentSeq = seq;
        messagePipe.Send(new OutgoingMessage(observerId, message, packetMode));
    }

    private void SendTeleport(PeerIndex observerId, ref PeerToPeerView view, PeerIndex subjectId, PeerSnapshot snapshot)
    {
        SendTracked(observerId, ref view, snapshot.Seq, new ServerMessage
        {
            Teleported = new TeleportPerformed
            {
                SubjectId = subjectId,
                Sequence = snapshot.Seq,
                ServerTick = snapshot.ServerTick,
                State = CreatePlayerState(snapshot),
                Realm = snapshot.Realm ?? string.Empty,
            },
        }, PacketMode.RELIABLE);

        logger.LogInformation("Broadcasting teleport from {Subject} to {ObserverId} at {Position}",
            subjectId, observerId, snapshot.GlobalPosition);
    }

    private void SendEmoteStarted(PeerIndex observerId, ref PeerToPeerView view, PeerIndex subjectId, PeerSnapshot snapshot, EmoteState emote,
        bool fromEviction = false)
    {
        var emoteStarted = new EmoteStarted
        {
            SubjectId = subjectId.Value,
            Sequence = snapshot.Seq,
            ServerTick = emote.StartTick,
            EmoteId = emote.EmoteId,
            PlayerState = CreatePlayerState(snapshot),
        };

        if (emote.Mask != null)
            emoteStarted.Mask = emote.Mask.Value;

        SendTracked(observerId, ref view, snapshot.Seq, new ServerMessage
        {
            EmoteStarted = emoteStarted,
        }, PacketMode.RELIABLE, fromEmoteStartEviction: fromEviction);

        logger.LogInformation("Broadcasting EmoteStarted {EmoteId} for subject {Subject} to observer {Observer}",
            emote.EmoteId, subjectId, observerId);
    }

    private void SendEmoteStopped(PeerIndex observerId, ref PeerToPeerView view, PeerIndex subjectId, PeerSnapshot snapshot, EmoteStopReason reason)
    {
        SendTracked(observerId, ref view, snapshot.Seq, new ServerMessage
        {
            EmoteStopped = new EmoteStopped
            {
                SubjectId = subjectId.Value,
                ServerTick = snapshot.ServerTick,
                Reason = reason,
                Sequence = snapshot.Seq,
                PlayerState = CreatePlayerState(snapshot),
            },
        }, PacketMode.RELIABLE);

        logger.LogInformation("Sending EmoteStopped for subject {Subject} to observer {Observer} (reason={Reason})",
            subjectId, observerId, reason);
    }

    private void SendDelta(PeerIndex observerId, ref PeerToPeerView view, PeerIndex subjectId, PeerSnapshot baseline, PeerSnapshot target,
        PeerViewSimulationTier tier,
        PacketMode packetMode,
        bool fromResync = false)
    {
        if (baseline.Seq == target.Seq)
            return;

        PlayerStateDeltaTier0 delta = PeerViewDiff.CreateMessage(subjectId, baseline, target, tier);

        SendTracked(observerId, ref view, target.Seq, new ServerMessage
        {
            PlayerStateDelta = delta,
        }, packetMode, fromResync: fromResync);
    }

    private void TryAnnounceProfile(PeerIndex observerId, PeerIndex subjectId, ref PeerToPeerView view)
    {
        int currentVersion = profileBoard.Get(subjectId);

        if (currentVersion != view.LastSentProfileVersion)
        {
            messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
            {
                PlayerProfileVersionAnnounced = new PlayerProfileVersionsAnnounced
                {
                    Version = currentVersion,
                    SubjectId = subjectId,
                },
            }, PacketMode.RELIABLE));

            logger.LogDebug("Profile version announced for subject {Subject} to observer {Observer} (v{PrevVersion} -> v{Version})",
                subjectId, observerId, view.LastSentProfileVersion, currentVersion);

            view.LastSentProfileVersion = currentVersion;
        }
    }

    // ── Cleanup ─────────────────────────────────────────────────────

    private void CleanupDisconnectedPeer(PeerIndex peerId)
    {
        snapshotBoard.ClearActive(peerId);
        spatialGrid.Remove(peerId);
        identityBoard.Remove(peerId);
        profileBoard.Remove(peerId);
        observerViews.Remove(peerId);
        peersToBeRemoved.Add(peerId);

        // Return the PeerIndex to the allocator last, after every per-peer board is wiped.
        // This is the one place in the system that unparks a slot — keeping cleanup and
        // reuse in lockstep.
        peerIndexAllocator.Release(peerId);

        logger.LogInformation("Peer {Peer} removed after disconnected", peerId);
    }

    /// <summary>
    ///     Periodic sweep — removes views not touched in recent ticks. Runs every <see cref="SWEEP_INTERVAL" /> ticks
    ///     to reclaim memory from subjects that left the interest set. Not on the hot path.
    /// </summary>
    private void SweepStaleViews(PeerIndex observerId, Dictionary<PeerIndex, PeerToPeerView> views, uint tickCounter)
    {
        if (views.Count == 0)
            return;

        sweepBuffer.Clear();

        foreach ((PeerIndex subjectId, PeerToPeerView view) in views)
        {
            if (tickCounter - view.LastSeenTick > SWEEP_INTERVAL)
                sweepBuffer.Add(subjectId);
        }

        foreach (PeerIndex id in sweepBuffer)
        {
            messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
            {
                PlayerLeft = new PlayerLeft { SubjectId = id },
            }, PacketMode.RELIABLE));

            logger.LogInformation("Sending PlayerLeft for subject {Subject} to observer {Observer} (stale view swept)", id, observerId);

            views.Remove(id);
        }
    }

    // ── State conversion ────────────────────────────────────────────

    private PlayerStateFull CreateFullState(PeerIndex subjectId, PeerSnapshot snapshot) =>
        new ()
        {
            SubjectId = subjectId.Value,
            Sequence = snapshot.Seq,
            ServerTick = snapshot.ServerTick,
            State = CreatePlayerState(snapshot),
        };

    private static PlayerState CreatePlayerState(PeerSnapshot snapshot)
    {
        // Snapshot already holds the raw quantized codes — copy them straight onto the outgoing
        // message, no re-encode.
        var state = new PlayerState
        {
            ParcelIndex = snapshot.Parcel,
            PositionX = snapshot.PositionX,
            PositionY = snapshot.PositionY,
            PositionZ = snapshot.PositionZ,
            VelocityX = snapshot.VelocityX,
            VelocityY = snapshot.VelocityY,
            VelocityZ = snapshot.VelocityZ,
            RotationY = snapshot.RotationY,
            MovementBlend = snapshot.MovementBlend,
            SlideBlend = snapshot.SlideBlend,
            StateFlags = (uint)snapshot.AnimationFlags,
            GlideState = snapshot.GlideState,
            JumpCount = snapshot.JumpCount,
        };

        if (snapshot.HeadYaw.HasValue)
            state.HeadYaw = snapshot.HeadYaw.Value;

        if (snapshot.HeadPitch.HasValue)
            state.HeadPitch = snapshot.HeadPitch.Value;

        if (snapshot.PointAt.HasValue)
        {
            QuantizedPointAt pointAt = snapshot.PointAt.Value;
            state.PointAtX = pointAt.X;
            state.PointAtY = pointAt.Y;
            state.PointAtZ = pointAt.Z;
        }

        return state;
    }
}