feat(velocity): wire whenOnline queue with player join and disconnect lifecycle

Author: objz

velocity/src/main/java/dev/objz/commandbridge/velocity/Main.java

@@ -160,6 +160,7 @@ public void onProxyInitialization(ProxyInitializeEvent e) {
         pluginMessageQueue = new PluginMessageQueue();
         sessions.onRemove(session -> {
             playerTracker.remove(session.id());
+            pluginMessageQueue.removeByServer(session.id());
             if (api != null) {
                 api.onServerDisconnected(session);
             }
@@ -211,6 +212,14 @@ public void onProxyInitialization(ProxyInitializeEvent e) {
 
         api = new VelocityCommandBridgeImpl(sessions, playerTracker, cfg.serverId(), endpointServer, pluginMessageQueue);
 
+        playerTracker.onPlayerJoin((clientId, uuid) -> {
+            if (api != null) {
+                for (var msg : pluginMessageQueue.drain(clientId, uuid)) {
+                    api.replayQueuedMessage(msg);
+                }
+            }
+        });
+
         installRoutes();
 
         command = new CBCommand(

velocity/src/main/java/dev/objz/commandbridge/velocity/api/PluginMessageQueue.java

@@ -21,7 +21,7 @@ public final class PluginMessageQueue {
     private static final long TTL_MS = TimeUnit.MINUTES.toMillis(5);
     private static final int MAX_TOTAL = 1000;
 
-    record QueuedMessage(Platform.ServerTarget target, PluginMessage message, String from, long queuedAt) {
+    public record QueuedMessage(Platform.ServerTarget target, PluginMessage message, String from, long queuedAt) {
     }
 
     private final ConcurrentHashMap<UUID, Queue<QueuedMessage>> queues = new ConcurrentHashMap<>();

velocity/src/main/java/dev/objz/commandbridge/velocity/api/VelocityCommandBridgeImpl.java

@@ -394,6 +394,27 @@ private void sendErrorResponse(Envelope originalEnv, String error) {
         });
     }
 
+    public void replayQueuedMessage(PluginMessageQueue.QueuedMessage msg) {
+        if (msg.from() == null) {
+            sendPluginMessage(msg.target(), msg.message()).exceptionally(ex -> {
+                Log.warn("Failed to replay queued plugin message: {}", ex.getMessage());
+                return null;
+            });
+            return;
+        }
+        Optional<ClientSession> targetSession = sessions.findSession(msg.target().id(),
+                toLocation(msg.target().type()));
+        if (targetSession.isEmpty() || !isRoutable(targetSession.get())) {
+            return;
+        }
+        Envelope env = Envelope.make(MessageType.PLUGIN_MESSAGE, msg.from(),
+                msg.target().id(), Envelope.MAPPER.valueToTree(msg.message()));
+        endpointServer.send(targetSession.get().endpoint(), env).dispatch().exceptionally(ex -> {
+            Log.warn("Failed to replay queued plugin message from {}: {}", msg.from(), ex.getMessage());
+            return null;
+        });
+    }
+
     private PluginMessage readPluginMessage(Envelope env) {
         try {
             return Envelope.MAPPER.treeToValue(env.payload(), PluginMessage.class);