feat: emotes (#7)

Author: lorux0

src/DCLPulse/Messaging/EmoteStartHandler.cs

@@ -0,0 +1,23 @@
+using Decentraland.Pulse;
+using Pulse.Peers;
+using Pulse.Peers.Simulation;
+
+namespace Pulse.Messaging;
+
+public class EmoteStartHandler(EmoteBoard emoteBoard, ITimeProvider timeProvider, ILogger<EmoteStartHandler> logger)
+    : RuntimePacketHandlerBase<EmoteStartHandler>(logger), IMessageHandler
+{
+    public void Handle(Dictionary<PeerIndex, PeerState> peers, PeerIndex from, ClientMessage message)
+    {
+        if (SkipFromUnauthorizedPeer(peers, from, message, out _))
+            return;
+
+        EmoteStart emoteStart = message.EmoteStart;
+
+        uint? durationMs = emoteStart.HasDurationMs ? emoteStart.DurationMs : null;
+
+        emoteBoard.Start(from, emoteStart.EmoteId, timeProvider.MonotonicTime, durationMs);
+
+        logger.LogDebug("Peer {Peer} started emote {EmoteId}", from.Value, emoteStart.EmoteId);
+    }
+}

src/DCLPulse/Messaging/EmoteStopHandler.cs

@@ -0,0 +1,25 @@
+using Decentraland.Pulse;
+using Pulse.Peers;
+using Pulse.Peers.Simulation;
+
+namespace Pulse.Messaging;
+
+public class EmoteStopHandler(EmoteBoard emoteBoard, ITimeProvider timeProvider, ILogger<EmoteStopHandler> logger)
+    : RuntimePacketHandlerBase<EmoteStopHandler>(logger), IMessageHandler
+{
+    public void Handle(Dictionary<PeerIndex, PeerState> peers, PeerIndex from, ClientMessage message)
+    {
+        if (SkipFromUnauthorizedPeer(peers, from, message, out _))
+            return;
+
+        if (!emoteBoard.IsEmoting(from))
+        {
+            logger.LogWarning("Peer {Peer} sent EmoteStop but no emote is active", from.Value);
+            return;
+        }
+
+        logger.LogDebug("Peer {Peer} stopped emote", from.Value);
+
+        emoteBoard.Stop(from, timeProvider.MonotonicTime);
+    }
+}

src/DCLPulse/Peers/PeerToPeerView.cs

@@ -34,4 +34,10 @@ public struct PeerToPeerView
     ///     Compared against <see cref="Simulation.ProfileBoard" /> each tick to detect changes.
     /// </summary>
     public int LastSentProfileVersion;
+
+    /// <summary>
+    ///     The emote ID last sent to the observer for this subject, or null if idle.
+    ///     Compared against <see cref="Simulation.EmoteBoard" /> each tick to detect transitions.
+    /// </summary>
+    public string? LastSentEmoteId;
 }

src/DCLPulse/Peers/PeersManager.cs

@@ -46,6 +46,7 @@ public sealed class PeersManager : BackgroundService
     private readonly Dictionary<ClientMessage.MessageOneofCase, IMessageHandler> messageHandlers;
     private readonly ITransport transport;
     private readonly ProfileBoard profileBoard;
+    private readonly EmoteBoard emoteBoard;
 
     public PeersManager(
         MessagePipe messagePipe,
@@ -60,7 +61,8 @@ public PeersManager(
         ITimeProvider timeProvider,
         Dictionary<ClientMessage.MessageOneofCase, IMessageHandler> messageHandlers,
         ITransport transport,
-        ProfileBoard profileBoard)
+        ProfileBoard profileBoard,
+        EmoteBoard emoteBoard)
     {
         this.messagePipe = messagePipe;
         this.logger = logger;
@@ -69,6 +71,7 @@ public PeersManager(
         this.messageHandlers = messageHandlers;
         this.transport = transport;
         this.profileBoard = profileBoard;
+        this.emoteBoard = emoteBoard;
         this.peerStateFactory = peerStateFactory;
         this.areaOfInterest = areaOfInterest;
         this.snapshotBoard = snapshotBoard;
@@ -103,7 +106,7 @@ protected override Task ExecuteAsync(CancellationToken stoppingToken)
         {
             var simulation = new PeerSimulation(
                 areaOfInterest, snapshotBoard, spatialGrid, identityBoard,
-                messagePipe, peerOptions.SimulationSteps, timeProvider, transport, profileBoard, peerSimulationLogger);
+                messagePipe, peerOptions.SimulationSteps, timeProvider, transport, profileBoard, emoteBoard, peerSimulationLogger);
 
             tasks[i + 1] = WorkerAsync(i, messageChannels[i].Reader,
                 peerLifeCycleChannels[i].Reader, simulation, stoppingToken);

src/DCLPulse/Peers/Simulation/EmoteBoard.cs

@@ -0,0 +1,44 @@
+using Decentraland.Pulse;
+
+namespace Pulse.Peers.Simulation;
+
+/// <summary>
+///     Shared emote state store indexed by <see cref="PeerIndex" />.
+///     Tracks which peer is currently emoting.
+///     <para />
+///     Writer: the handler on whichever worker owns the peer.
+///     Readers: simulation step on any worker, diffs against <see cref="PeerToPeerView.LastSentEmoteId" />.
+/// </summary>
+public class EmoteBoard(int maxPeers)
+{
+    private readonly EmoteState?[] states = new EmoteState?[maxPeers];
+
+    public void Start(PeerIndex id, string emoteId, uint serverTick, uint? durationMs) =>
+        Volatile.Write(ref states[(int)id.Value], new EmoteState(emoteId, serverTick, null, durationMs));
+
+    public void Stop(PeerIndex id, uint serverTick) =>
+        // Maybe it could be useful to keep the start tick if its needed somehow in the future
+        Volatile.Write(ref states[(int)id.Value], new EmoteState(null, 0, serverTick, StopReason: EmoteStopReason.Cancelled));
+
+    public EmoteState? Get(PeerIndex id) =>
+        Volatile.Read(ref states[(int)id.Value]);
+
+    public bool IsEmoting(PeerIndex id) =>
+        Volatile.Read(ref states[(int)id.Value])?.EmoteId != null;
+
+    public void TryComplete(PeerIndex id, uint now)
+    {
+        EmoteState? state = Volatile.Read(ref states[(int)id.Value]);
+
+        if (state?.EmoteId == null || state.DurationMs == null)
+            return;
+
+        if (now - state.StartTick >= state.DurationMs.Value)
+            Volatile.Write(ref states[(int)id.Value], new EmoteState(null, 0, now, StopReason: EmoteStopReason.Completed));
+    }
+
+    public void Remove(PeerIndex id) =>
+        Volatile.Write(ref states[(int)id.Value], null);
+}
+
+public record EmoteState(string? EmoteId, uint StartTick, uint? StopTick, uint? DurationMs = null, EmoteStopReason? StopReason = null);

src/DCLPulse/Peers/Simulation/PeerSimulation.cs

@@ -31,6 +31,7 @@ public sealed class PeerSimulation : IPeerSimulation
     private readonly ITimeProvider timeProvider;
     private readonly ITransport transport;
     private readonly ProfileBoard profileBoard;
+    private readonly EmoteBoard emoteBoard;
     private readonly ILogger<PeerSimulation> logger;
 
     /// <summary>
@@ -69,6 +70,7 @@ public PeerSimulation(
         ITimeProvider timeProvider,
         ITransport transport,
         ProfileBoard profileBoard,
+        EmoteBoard emoteBoard,
         ILogger<PeerSimulation> logger)
     {
         this.areaOfInterest = areaOfInterest;
@@ -80,6 +82,7 @@ public PeerSimulation(
         this.timeProvider = timeProvider;
         this.transport = transport;
         this.profileBoard = profileBoard;
+        this.emoteBoard = emoteBoard;
         this.logger = logger;
 
         BaseTickMs = simulationSteps[0];
@@ -116,6 +119,7 @@ public void SimulateTick(Dictionary<PeerIndex, PeerState> peers, uint tickCounte
                     spatialGrid.Remove(observerId);
                     identityBoard.Remove(observerId);
                     profileBoard.Remove(observerId);
+                    emoteBoard.Remove(observerId);
                     observerViews.Remove(observerId);
                     peersToBeRemoved.Add(observerId);
                     logger.LogInformation("Peer {Peer} removed after disconnected", observerId);
@@ -126,6 +130,9 @@ public void SimulateTick(Dictionary<PeerIndex, PeerState> peers, uint tickCounte
             if (observerState.ConnectionState != PeerConnectionState.AUTHENTICATED)
                 continue;
 
+            // Completes the emote in case no observer is near this peer
+            emoteBoard.TryComplete(observerId, timeProvider.MonotonicTime);
+
             if (!snapshotBoard.TryRead(observerId, out PeerSnapshot observerSnapshot))
                 continue;
 
@@ -236,9 +243,12 @@ private void ProcessVisibleSubjects(
                 else
                     SendDelta(view.LastSentSnapshot, ITransport.PacketMode.UNRELIABLE_SEQUENCED);
 
+                // Only announce on delta because PlayerJoined is considered as an announcement
                 TryAnnounceProfile();
             }
 
+            SyncEmoteState();
+
             view.LastSentSeq = subjectSnapshot.Seq;
             view.LastSentSnapshot = subjectSnapshot;
             view.LastSeenTick = tickCounter;
@@ -265,6 +275,46 @@ void TryAnnounceProfile()
                 }
             }
 
+            void SyncEmoteState()
+            {
+                // If the emote completion has not been process by the peer's worker yet, try to complete it now
+                emoteBoard.TryComplete(entry.Subject, timeProvider.MonotonicTime);
+
+                EmoteState? emoteState = emoteBoard.Get(entry.Subject);
+                string? currentEmote = emoteState?.EmoteId;
+
+                if (currentEmote == view.LastSentEmoteId)
+                    return;
+
+                if (currentEmote != null)
+                {
+                    messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
+                    {
+                        EmoteStarted = new EmoteStarted
+                        {
+                            SubjectId = entry.Subject.Value,
+                            ServerTick = emoteState!.StartTick,
+                            EmoteId = currentEmote,
+                            PlayerState = CreatePlayerState(subjectSnapshot),
+                        },
+                    }, ITransport.PacketMode.RELIABLE));
+                }
+                else if (emoteState is { StopTick: not null, StopReason: not null })
+                {
+                    messagePipe.Send(new OutgoingMessage(observerId, new ServerMessage
+                    {
+                        EmoteStopped = new EmoteStopped
+                        {
+                            SubjectId = entry.Subject.Value,
+                            ServerTick = emoteState.StopTick.Value,
+                            Reason = emoteState.StopReason.Value,
+                        },
+                    }, ITransport.PacketMode.RELIABLE));
+                }
+
+                view.LastSentEmoteId = currentEmote;
+            }
+
             bool SendDelta(PeerSnapshot baseline, ITransport.PacketMode packetMode)
             {
                 if (baseline.Seq == subjectSnapshot.Seq)
@@ -313,7 +363,16 @@ private void SweepStaleViews(PeerIndex observerId, Dictionary<PeerIndex, PeerToP
         }
     }
 
-    private PlayerStateFull CreateFullState(PeerIndex subjectId, PeerSnapshot snapshot)
+    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)
     {
         var state = new PlayerState
         {
@@ -333,12 +392,6 @@ private PlayerStateFull CreateFullState(PeerIndex subjectId, PeerSnapshot snapsh
         if (snapshot.HeadPitch.HasValue)
             state.HeadPitch = snapshot.HeadPitch.Value;
 
-        return new PlayerStateFull
-        {
-            SubjectId = subjectId.Value,
-            Sequence = snapshot.Seq,
-            ServerTick = snapshot.ServerTick,
-            State = state,
-        };
+        return state;
     }
 }

src/DCLPulse/Program.cs

@@ -31,6 +31,8 @@
 builder.Services.AddSingleton<ResyncRequestHandler>();
 builder.Services.AddSingleton<HandshakeHandler>();
 builder.Services.AddSingleton<ProfileAnnouncementHandler>();
+builder.Services.AddSingleton<EmoteStartHandler>();
+builder.Services.AddSingleton<EmoteStopHandler>();
 builder.Services.AddSingleton(new AuthChainValidator(new NethereumPersonalSignVerifier()));
 
 builder.Services.AddSingleton(sp => new Dictionary<ClientMessage.MessageOneofCase, IMessageHandler>
@@ -39,6 +41,8 @@
     { ClientMessage.MessageOneofCase.Input, sp.GetRequiredService<PlayerStateInputHandler>() },
     { ClientMessage.MessageOneofCase.Resync, sp.GetRequiredService<ResyncRequestHandler>() },
     { ClientMessage.MessageOneofCase.ProfileAnnouncement, sp.GetRequiredService<ProfileAnnouncementHandler>() },
+    { ClientMessage.MessageOneofCase.EmoteStart, sp.GetRequiredService<EmoteStartHandler>() },
+    { ClientMessage.MessageOneofCase.EmoteStop, sp.GetRequiredService<EmoteStopHandler>() },
 });
 
 builder.Services.AddSingleton<ProfileBoard>(sp =>
@@ -47,6 +51,12 @@
     return new ProfileBoard(transportOptions.MaxPeers);
 });
 
+builder.Services.AddSingleton(sp =>
+{
+    ENetTransportOptions transportOptions = sp.GetRequiredService<IOptions<ENetTransportOptions>>().Value;
+    return new EmoteBoard(transportOptions.MaxPeers);
+});
+
 // Simulation
 builder.Services.AddSingleton(sp =>
 {

src/DCLPulseTests/DrainPeerLifeCycleEventsTests.cs

@@ -41,7 +41,8 @@ public void SetUp()
             timeProvider,
             new Dictionary<ClientMessage.MessageOneofCase, IMessageHandler>(),
             Substitute.For<ITransport>(),
-            new ProfileBoard(100));
+            new ProfileBoard(100),
+            new EmoteBoard(100));
 
         lifeCycleChannel = Channel.CreateUnbounded<MessagePipe.PeerLifeCycleEvent>();
         peers = new Dictionary<PeerIndex, PeerState>();