Add player presence reconciliation for proxy servers

api/src/main/java/app/simplecloud/api/CloudApi.java

@@ -38,7 +38,7 @@
  * });
  * }</pre>
  */
-public interface CloudApi {
+public interface CloudApi extends AutoCloseable {
 
     /**
      * Creates a CloudAPI instance with default options.
@@ -143,4 +143,8 @@ static CloudApi create(CloudApiOptions options) {
      */
     QueryCache cache();
 
+    @Override
+    default void close() {
+    }
+
 }

api/src/main/java/app/simplecloud/api/internal/CloudApiImpl.java

@@ -21,6 +21,7 @@
 import io.nats.client.Connection;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class CloudApiImpl implements CloudApi {
 
@@ -35,6 +36,7 @@ public class CloudApiImpl implements CloudApi {
     private final PersistentServerApi persistentServerApi;
     private final EventApi eventApi;
     private final PlayerApi playerApi;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     public CloudApiImpl(CloudApiOptions options) {
         this.options = options;
@@ -73,6 +75,7 @@ public CloudApiImpl(CloudApiOptions options) {
         } else {
             this.cacheEventListener = null;
         }
+
     }
 
     @Override
@@ -120,4 +123,19 @@ public QueryCache cache() {
         return queryCache;
     }
 
+    @Override
+    public void close() {
+        if (!closed.compareAndSet(false, true)) {
+            return;
+        }
+
+        if (cacheEventListener != null) {
+            cacheEventListener.shutdown();
+        }
+        if (queryCache instanceof QueryCacheImpl queryCacheImpl) {
+            queryCacheImpl.shutdown();
+        }
+        natsConnectionManager.shutdown();
+    }
+
 }

api/src/main/java/app/simplecloud/api/internal/integration/player/PlayerIntegration.java

@@ -1,6 +1,5 @@
 package app.simplecloud.api.internal.integration.player;
 
-import app.simplecloud.api.CloudApi;
 import app.simplecloud.api.internal.CloudApiImpl;
 import app.simplecloud.api.player.CloudPlayer;
 import build.buf.gen.simplecloud.player.v2.*;
@@ -29,12 +28,8 @@ public class PlayerIntegration {
     private BiFunction<String, String, CompletableFuture<Boolean>> kickHandler;
     private BiFunction<String, String, CompletableFuture<CloudPlayer.ConnectResult>> connectHandler;
 
-    public PlayerIntegration(CloudApi cloudApi) {
-        if (!(cloudApi instanceof CloudApiImpl)) {
-            throw new IllegalArgumentException("CloudApi must be an instance of CloudApiImpl");
-        }
-        CloudApiImpl impl = (CloudApiImpl) cloudApi;
-        this.natsConnection = impl.getNatsConnection();
+    public PlayerIntegration(CloudApiImpl cloudApi) {
+        this.natsConnection = cloudApi.getNatsConnection();
         this.networkId = cloudApi.getNetworkId();
     }
 

api/src/main/java/app/simplecloud/api/internal/integration/presence/ProxyPresenceResponder.java

@@ -0,0 +1,149 @@
+package app.simplecloud.api.internal.integration.presence;
+
+import app.simplecloud.api.presence.ProxyPresencePlayer;
+import app.simplecloud.api.presence.ProxyPresencePlayerProvider;
+import build.buf.gen.simplecloud.controller.v2.PresenceCompareRequest;
+import build.buf.gen.simplecloud.controller.v2.ProxyPresenceCompareResponse;
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import io.nats.client.Message;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Responds to controller presence-compare requests for a single proxy.
+ */
+public final class ProxyPresenceResponder {
+
+    private static final Logger LOGGER = Logger.getLogger(ProxyPresenceResponder.class.getName());
+    private static final int FNV_32A_OFFSET_BASIS = 0x811c9dc5;
+    private static final int FNV_32A_PRIME = 0x01000193;
+
+    private final Connection natsConnection;
+    private final String serverId;
+    private final String subject;
+    private final AtomicBoolean running = new AtomicBoolean(false);
+    private volatile ProxyPresencePlayerProvider playerProvider;
+
+    private Dispatcher dispatcher;
+
+    public ProxyPresenceResponder(
+            Connection natsConnection,
+            String networkId,
+            String serverId,
+            ProxyPresencePlayerProvider playerProvider
+    ) {
+        this.natsConnection = Objects.requireNonNull(natsConnection, "natsConnection");
+        this.serverId = serverId == null ? "" : serverId;
+        this.subject = Objects.requireNonNull(networkId, "networkId") + ".server." + this.serverId + ".presence.compare";
+        this.playerProvider = playerProvider;
+    }
+
+    public ProxyPresenceResponder(
+            Connection natsConnection,
+            String networkId,
+            String serverId
+    ) {
+        this(natsConnection, networkId, serverId, null);
+    }
+
+    public void start() {
+        if (serverId.isBlank()) {
+            LOGGER.warning("Presence responder not started because SIMPLECLOUD_UNIQUE_ID is missing");
+            return;
+        }
+        if (!running.compareAndSet(false, true)) {
+            return;
+        }
+
+        dispatcher = natsConnection.createDispatcher(null);
+        dispatcher.subscribe(subject, this::handleCompareRequest);
+    }
+
+    public void registerPlayerProvider(ProxyPresencePlayerProvider playerProvider) {
+        this.playerProvider = Objects.requireNonNull(playerProvider, "playerProvider");
+    }
+
+    public void unregisterPlayerProvider() {
+        this.playerProvider = null;
+    }
+
+    public void stop() {
+        if (!running.compareAndSet(true, false)) {
+            return;
+        }
+        if (dispatcher != null) {
+            dispatcher.unsubscribe(subject);
+        }
+    }
+
+    private void handleCompareRequest(Message message) {
+        String replyTo = message.getReplyTo();
+        if (replyTo == null || replyTo.isBlank()) {
+            return;
+        }
+
+        try {
+            PresenceCompareRequest request = PresenceCompareRequest.parseFrom(message.getData());
+            List<ProxyPresencePlayer> players = currentPlayers();
+            int localHash = computeHash(players);
+            boolean match = localHash == request.getHash();
+
+            LOGGER.info("[Presence] Compare request received — controller hash: " + request.getHash()
+                    + ", local hash: " + localHash + ", match: " + match
+                    + ", online players (" + players.size() + "): "
+                    + players.stream().map(p -> p.getName() + "(" + p.getPlayerId() + ")").toList());
+
+            ProxyPresenceCompareResponse.Builder response = ProxyPresenceCompareResponse.newBuilder()
+                    .setMatch(match);
+
+            if (!match) {
+                response.addAllPlayers(players.stream().map(ProxyPresencePlayer::toProto).toList());
+            }
+
+            natsConnection.publish(replyTo, response.build().toByteArray());
+        } catch (Exception e) {
+            LOGGER.log(Level.WARNING, "Failed to process presence compare request for " + subject, e);
+        }
+    }
+
+    private List<ProxyPresencePlayer> currentPlayers() {
+        ProxyPresencePlayerProvider currentProvider = playerProvider;
+        if (currentProvider == null) {
+            return List.of();
+        }
+
+        Collection<ProxyPresencePlayer> suppliedPlayers = currentProvider.getProxyPresencePlayers();
+        if (suppliedPlayers == null || suppliedPlayers.isEmpty()) {
+            return List.of();
+        }
+
+        return suppliedPlayers.stream()
+                .filter(Objects::nonNull)
+                .sorted(Comparator.comparing(ProxyPresencePlayer::hashRecord))
+                .toList();
+    }
+
+    static int computeHash(Collection<ProxyPresencePlayer> players) {
+        if (players == null || players.isEmpty()) {
+            return 0;
+        }
+
+        int hash = FNV_32A_OFFSET_BASIS;
+        for (ProxyPresencePlayer player : players) {
+            byte[] bytes = player.hashRecord().getBytes(StandardCharsets.UTF_8);
+            for (byte currentByte : bytes) {
+                hash ^= currentByte & 0xff;
+                hash *= FNV_32A_PRIME;
+            }
+        }
+        return hash;
+    }
+}

api/src/main/java/app/simplecloud/api/internal/integration/presence/ProxyPresenceTracker.java

@@ -0,0 +1,86 @@
+package app.simplecloud.api.internal.integration.presence;
+
+import app.simplecloud.api.presence.ProxyPresencePlayer;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Tracks stable per-player metadata needed for proxy presence reconciliation.
+ */
+public final class ProxyPresenceTracker {
+
+    private final String connectedProxyName;
+    private final Map<String, TrackedPlayerMetadata> players = new ConcurrentHashMap<>();
+
+    public ProxyPresenceTracker(String connectedProxyName) {
+        this.connectedProxyName = connectedProxyName == null ? "" : connectedProxyName;
+    }
+
+    public void trackLogin(String playerId) {
+        players.put(normalize(playerId), new TrackedPlayerMetadata(System.currentTimeMillis()));
+    }
+
+    public void updateSessionId(String playerId, String sessionId) {
+        TrackedPlayerMetadata metadata = players.get(normalize(playerId));
+        if (metadata != null) {
+            metadata.setSessionId(sessionId);
+        }
+    }
+
+    public void remove(String playerId) {
+        players.remove(normalize(playerId));
+    }
+
+    public ProxyPresencePlayer createSnapshot(
+            String playerId,
+            String name,
+            String displayName,
+            String connectedServerName,
+            String clientLanguage,
+            int clientVersion,
+            boolean onlineMode
+    ) {
+        String normalizedPlayerId = normalize(playerId);
+        TrackedPlayerMetadata metadata = players.get(normalizedPlayerId);
+
+        return new ProxyPresencePlayer(
+                normalizedPlayerId,
+                name,
+                displayName,
+                connectedServerName,
+                connectedProxyName,
+                metadata != null ? metadata.getLoginTimestampUnixMillis() : 0L,
+                clientLanguage,
+                clientVersion,
+                onlineMode,
+                metadata != null ? metadata.getSessionId() : ""
+        );
+    }
+
+    private static String normalize(String value) {
+        return value == null ? "" : value;
+    }
+
+    private static final class TrackedPlayerMetadata {
+
+        private final long loginTimestampUnixMillis;
+        private volatile String sessionId = "";
+
+        private TrackedPlayerMetadata(long loginTimestampUnixMillis) {
+            this.loginTimestampUnixMillis = loginTimestampUnixMillis;
+        }
+
+        private long getLoginTimestampUnixMillis() {
+            return loginTimestampUnixMillis;
+        }
+
+        private String getSessionId() {
+            return sessionId;
+        }
+
+        private void setSessionId(String sessionId) {
+            this.sessionId = sessionId == null ? "" : sessionId;
+        }
+    }
+}