Resend player meta when LuckPerms permissions change

src/main/java/net/tfminecraft/tfmcweb/TFMCWeb.java

@@ -11,6 +11,7 @@
 import net.tfminecraft.tfmcweb.listeners.PlayerJoinListener;
 import net.tfminecraft.tfmcweb.loaders.ConfigLoader;
 import net.tfminecraft.tfmcweb.luckperms.LuckPermsBridge;
+import net.tfminecraft.tfmcweb.entitlements.MetaResync;
 import net.tfminecraft.tfmcweb.entitlements.PlayerMetaSyncService;
 import net.tfminecraft.tfmcweb.managers.BanMirrorPoller;
 import net.tfminecraft.tfmcweb.managers.LinkDiscordCommand;
@@ -33,6 +34,7 @@ public class TFMCWeb extends JavaPlugin {
 	private PatreonRankWriter rankWriter;
 	private LuckPermsBridge luckPermsBridge;
 	private BanMirrorPoller banMirror;
+	private MetaResync metaResync;
 
 	@Override
 	public void onEnable() {
@@ -108,6 +110,8 @@ public void onEnable() {
 		);
 		banMirror = new BanMirrorPoller(this, linkCache);
 		banMirror.start();
+		metaResync = new MetaResync(this);
+		metaResync.start();
 
 		getLogger().info(
 			"TFMCWeb enabled (api="
@@ -132,6 +136,9 @@ public void onDisable() {
 		if (banMirror != null) {
 			banMirror.stop();
 		}
+		if (metaResync != null) {
+			metaResync.stop();
+		}
 	}
 
 	public LinkCache getLinkCache() {

src/main/java/net/tfminecraft/tfmcweb/entitlements/MetaResync.java

@@ -0,0 +1,134 @@
+package net.tfminecraft.tfmcweb.entitlements;
+
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.logging.Level;
+
+import org.bukkit.Bukkit;
+import org.bukkit.entity.Player;
+import org.bukkit.plugin.IllegalPluginAccessException;
+import org.bukkit.plugin.Plugin;
+import org.bukkit.plugin.java.JavaPlugin;
+
+/**
+ * Pushes web entitlements again after LuckPerms recalculates a player's or a group's
+ * permissions, so rank changes made while players are online reach the website without a rejoin.
+ * Recalculations within one second share one check, and unchanged entitlements are not resent:
+ * LuckPerms recalculates every online player and group on each periodic storage sync.
+ */
+public final class MetaResync {
+
+	static final long DELAY_TICKS = 20L;
+
+	@FunctionalInterface
+	public interface Subscriber {
+		AutoCloseable subscribe(Plugin plugin, Consumer<UUID> onUser, Runnable onGroup);
+	}
+
+	private final JavaPlugin plugin;
+	private final Subscriber subscriber;
+	private final Set<UUID> pending = ConcurrentHashMap.newKeySet();
+	private final AtomicBoolean allPending = new AtomicBoolean();
+	private AutoCloseable subscription;
+
+	public MetaResync(JavaPlugin plugin) {
+		this(plugin, MetaResync::subscribeLuckPerms);
+	}
+
+	MetaResync(JavaPlugin plugin, Subscriber subscriber) {
+		this.plugin = plugin;
+		this.subscriber = subscriber;
+	}
+
+	/** Loads the LuckPerms listener by name so this class does not reference the API. */
+	static AutoCloseable subscribeLuckPerms(Plugin plugin, Consumer<UUID> onUser, Runnable onGroup) {
+		try {
+			Class<?> type = Class.forName("net.tfminecraft.tfmcweb.luckperms.LuckPermsRecalculateListener");
+			return (AutoCloseable) type.getMethod("subscribe", Plugin.class, Consumer.class, Runnable.class)
+				.invoke(null, plugin, onUser, onGroup);
+		} catch (ReflectiveOperationException | NoClassDefFoundError e) {
+			throw new IllegalStateException("LuckPerms API is not available", e);
+		}
+	}
+
+	public void start() {
+		if (subscription != null || !luckPermsPresent()) {
+			return;
+		}
+		try {
+			subscription = subscriber.subscribe(plugin, this::request, this::requestAll);
+		} catch (RuntimeException e) {
+			plugin.getLogger().log(
+				Level.WARNING,
+				"[player-meta] LuckPerms is unavailable; rank changes reach the website on next join",
+				e
+			);
+		}
+	}
+
+	public void stop() {
+		pending.clear();
+		allPending.set(false);
+		if (subscription == null) {
+			return;
+		}
+		try {
+			subscription.close();
+		} catch (Exception e) {
+			plugin.getLogger().log(Level.WARNING, "[player-meta] could not unsubscribe from LuckPerms", e);
+		}
+		subscription = null;
+	}
+
+	/** Safe from any thread; LuckPerms fires recalculations asynchronously. */
+	void request(UUID playerUuid) {
+		if (playerUuid == null || !plugin.isEnabled() || !pending.add(playerUuid)) {
+			return;
+		}
+		if (!schedule(() -> {
+			pending.remove(playerUuid);
+			Player online = Bukkit.getPlayer(playerUuid);
+			if (online != null && online.isOnline()) {
+				PlayerMetaSyncService.pushIfChanged(online);
+			}
+		})) {
+			pending.remove(playerUuid);
+		}
+	}
+
+	/** Checks every online player; a group change can reach any member. */
+	void requestAll() {
+		if (!plugin.isEnabled() || !allPending.compareAndSet(false, true)) {
+			return;
+		}
+		if (!schedule(() -> {
+			allPending.set(false);
+			for (Player online : Bukkit.getOnlinePlayers()) {
+				PlayerMetaSyncService.pushIfChanged(online);
+			}
+		})) {
+			allPending.set(false);
+		}
+	}
+
+	/** False when the plugin was disabled between the check and scheduling. */
+	private boolean schedule(Runnable task) {
+		try {
+			Bukkit.getScheduler().runTaskLater(plugin, task, DELAY_TICKS);
+			return true;
+		} catch (IllegalPluginAccessException e) {
+			return false;
+		}
+	}
+
+	private static boolean luckPermsPresent() {
+		if (Bukkit.getPluginManager() == null) {
+			return false;
+		}
+		Plugin luckPerms = Bukkit.getPluginManager().getPlugin("LuckPerms");
+		return luckPerms != null && luckPerms.isEnabled();
+	}
+}

src/main/java/net/tfminecraft/tfmcweb/entitlements/PlayerMetaSyncService.java

@@ -1,7 +1,10 @@
 package net.tfminecraft.tfmcweb.entitlements;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongSupplier;
 
 import org.bukkit.Bukkit;
 import org.bukkit.entity.Player;
@@ -12,30 +15,88 @@
 import net.tfminecraft.tfmcweb.entitlements.EntitlementResolver.ResolvedMeta;
 
 /**
- * Push resolved web entitlements to ProvinceSystem on join / reload (fail-soft).
+ * Push resolved web entitlements to ProvinceSystem on join, reload and rank change (fail-soft).
  */
 public final class PlayerMetaSyncService {
 
+	/**
+	 * After a failed PUT, unchanged bodies keep being resent until one succeeds at least this long
+	 * after the last failure. A timed-out PUT can still be applied by the website after a newer
+	 * one; resends on later recalculations (the periodic LuckPerms sync) repair that, best effort.
+	 */
+	static final long REPAIR_WINDOW_MS = 10 * 60_000L;
+	static LongSupplier clock = System::currentTimeMillis;
+
+	/** Newest body handed to the sender per player. Sends are attempted one at a time, in order. */
+	private static final Map<UUID, String> latest = new ConcurrentHashMap<>();
+	/** Time of each player's last failed PUT, cleared by a successful send after the window. */
+	private static final Map<UUID, Long> failedAt = new ConcurrentHashMap<>();
+	/** Newest body waiting to be sent per player. */
+	private static final Map<UUID, String> queued = new ConcurrentHashMap<>();
+	/** Players with a send task running; one per player keeps PUTs in order. */
+	private static final Set<UUID> sending = ConcurrentHashMap.newKeySet();
+
 	private PlayerMetaSyncService() {}
 
-	/** Resolve on the calling thread (prefer main), then PUT async. */
+	/** Resolve on the main thread, then PUT async. Producers must stay on the main thread. */
 	public static void pushForPlayer(Player player) {
+		push(player, true);
+	}
+
+	/** Like {@link #pushForPlayer} but skips the PUT when this body was already sent or is on its way. */
+	public static void pushIfChanged(Player player) {
+		push(player, false);
+	}
+
+	private static void push(Player player, boolean force) {
 		if (player == null || TFMCWeb.plugin == null) {
 			return;
 		}
 		UUID playerUuid = player.getUniqueId();
 		ResolvedMeta meta = EntitlementResolver.resolve(player);
 		String body = toJson(playerUuid, meta);
-		Bukkit.getScheduler().runTaskAsynchronously(
-			TFMCWeb.plugin,
-			() -> {
-				SimpleResult result = ProvinceSystemClient.putRpcPlayerMeta(body);
-				if (!result.ok) {
-					TFMCWeb.plugin.getLogger().warning(
-						"[player-meta] push failed for " + playerUuid + ": " + result.error
-					);
-				}
+		if (!force && body.equals(latest.get(playerUuid)) && !failedAt.containsKey(playerUuid)) {
+			return;
+		}
+		latest.put(playerUuid, body);
+		queued.put(playerUuid, body);
+		if (sending.add(playerUuid)) {
+			Bukkit.getScheduler().runTaskAsynchronously(TFMCWeb.plugin, () -> drain(playerUuid));
+		}
+	}
+
+	/**
+	 * Sends the newest queued body until none is left, one attempt at a time.
+	 * The outer check catches a body queued after the inner loop emptied but before release.
+	 */
+	static void drain(UUID playerUuid) {
+		do {
+			String body;
+			while ((body = queued.remove(playerUuid)) != null) {
+				send(playerUuid, body);
 			}
+			sending.remove(playerUuid);
+		} while (queued.containsKey(playerUuid) && sending.add(playerUuid));
+	}
+
+	private static void send(UUID playerUuid, String body) {
+		SimpleResult result;
+		try {
+			result = ProvinceSystemClient.putRpcPlayerMeta(body);
+		} catch (RuntimeException e) {
+			// A stuck sending entry would stop this player's pushes until restart.
+			result = SimpleResult.fail(String.valueOf(e));
+		}
+		if (result.ok) {
+			Long failed = failedAt.get(playerUuid);
+			if (failed != null && clock.getAsLong() - failed >= REPAIR_WINDOW_MS) {
+				failedAt.remove(playerUuid, failed);
+			}
+			return;
+		}
+		failedAt.put(playerUuid, clock.getAsLong());
+		TFMCWeb.plugin.getLogger().warning(
+			"[player-meta] push failed for " + playerUuid + ": " + result.error
 		);
 	}
 

src/main/java/net/tfminecraft/tfmcweb/luckperms/LuckPermsRecalculateListener.java

@@ -0,0 +1,39 @@
+package net.tfminecraft.tfmcweb.luckperms;
+
+import java.util.UUID;
+import java.util.function.Consumer;
+
+import org.bukkit.plugin.Plugin;
+
+import net.luckperms.api.LuckPermsProvider;
+import net.luckperms.api.event.EventBus;
+import net.luckperms.api.event.EventSubscription;
+import net.luckperms.api.event.group.GroupDataRecalculateEvent;
+import net.luckperms.api.event.user.UserDataRecalculateEvent;
+
+/**
+ * Reports LuckPerms permission recalculations: one player's, or a group's, which can change
+ * any member's permissions. Loaded by name so LuckPerms stays optional.
+ */
+public final class LuckPermsRecalculateListener {
+
+	private LuckPermsRecalculateListener() {}
+
+	public static AutoCloseable subscribe(Plugin plugin, Consumer<UUID> onUser, Runnable onGroup) {
+		EventBus events = LuckPermsProvider.get().getEventBus();
+		EventSubscription<UserDataRecalculateEvent> users = events.subscribe(
+			plugin,
+			UserDataRecalculateEvent.class,
+			event -> onUser.accept(event.getUser().getUniqueId())
+		);
+		EventSubscription<GroupDataRecalculateEvent> groups = events.subscribe(
+			plugin,
+			GroupDataRecalculateEvent.class,
+			event -> onGroup.run()
+		);
+		return () -> {
+			users.close();
+			groups.close();
+		};
+	}
+}

src/test/java/net/tfminecraft/tfmcweb/entitlements/EntitlementsTest.java

@@ -85,6 +85,57 @@ MetaGroupDefinition group(String permission, int tier, Map<String,Integer> perks
             PlayerMetaSyncService.pushAllOnlineAsync();assertEquals(1,async.size());async.removeFirst().run();verify(logger).warning(contains("unavailable"));
         }
     }
+    @Test void changeOnlySyncSkipsRepeatsAndRetriesAfterFailure() {
+        UUID id=UUID.randomUUID();Player player=mock(Player.class);when(player.getUniqueId()).thenReturn(id);Cache.playerMetaSyncPermissions=List.of("tfmc.flag");
+        TFMCWeb plugin=mock(TFMCWeb.class);when(plugin.getLogger()).thenReturn(mock(java.util.logging.Logger.class));TFMCWeb.plugin=plugin;
+        BukkitScheduler scheduler=mock(BukkitScheduler.class);List<Runnable> async=new ArrayList<>();
+        doAnswer(call->{async.add(call.getArgument(1,Runnable.class));return null;}).when(scheduler).runTaskAsynchronously(eq(plugin),any(Runnable.class));
+        try(MockedStatic<Bukkit> b=mockStatic(Bukkit.class);MockedStatic<ProvinceSystemClient> api=mockStatic(ProvinceSystemClient.class)) {
+            b.when(Bukkit::getScheduler).thenReturn(scheduler);
+            api.when(()->ProvinceSystemClient.putRpcPlayerMeta(anyString())).thenReturn(ProvinceSystemClient.SimpleResult.success());
+            PlayerMetaSyncService.pushIfChanged(null);PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());async.removeFirst().run();
+            PlayerMetaSyncService.pushIfChanged(player);assertTrue(async.isEmpty());
+            PlayerMetaSyncService.pushForPlayer(player);assertEquals(1,async.size());async.removeFirst().run();
+            when(player.hasPermission("tfmc.flag")).thenReturn(true);PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());
+            api.when(()->ProvinceSystemClient.putRpcPlayerMeta(anyString())).thenReturn(ProvinceSystemClient.SimpleResult.fail("down"));async.removeFirst().run();
+            PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());
+            api.when(()->ProvinceSystemClient.putRpcPlayerMeta(anyString())).thenThrow(new IllegalStateException("boom"));async.removeFirst().run();
+            verify(plugin.getLogger()).warning(contains("boom"));
+            PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());
+            api.verify(()->ProvinceSystemClient.putRpcPlayerMeta(contains("\"tfmc.flag\":true")),times(2));
+        }
+    }
+    @Test void aLateTimedOutWriteIsRepairedAndTheWindowEndsOnlyAfterASuccessPastIt() {
+        UUID id=UUID.randomUUID();Player player=mock(Player.class);when(player.getUniqueId()).thenReturn(id);Cache.playerMetaSyncPermissions=List.of("tfmc.flag");
+        TFMCWeb plugin=mock(TFMCWeb.class);when(plugin.getLogger()).thenReturn(mock(java.util.logging.Logger.class));TFMCWeb.plugin=plugin;
+        BukkitScheduler scheduler=mock(BukkitScheduler.class);List<Runnable> async=new ArrayList<>();
+        doAnswer(call->{async.add(call.getArgument(1,Runnable.class));return null;}).when(scheduler).runTaskAsynchronously(eq(plugin),any(Runnable.class));
+        long[] now={1_000L};var original=PlayerMetaSyncService.clock;PlayerMetaSyncService.clock=()->now[0];
+        String[] site={null};String[] lateWrite={null};boolean[] timeOut={true};
+        try(MockedStatic<Bukkit> b=mockStatic(Bukkit.class);MockedStatic<ProvinceSystemClient> api=mockStatic(ProvinceSystemClient.class)) {
+            b.when(Bukkit::getScheduler).thenReturn(scheduler);
+            // Fake site: a timed-out PUT is applied later, when the test says so.
+            api.when(()->ProvinceSystemClient.putRpcPlayerMeta(anyString())).thenAnswer(c->{
+                String body=c.getArgument(0);
+                if(timeOut[0]){timeOut[0]=false;lateWrite[0]=body;return ProvinceSystemClient.SimpleResult.fail("timeout");}
+                site[0]=body;return ProvinceSystemClient.SimpleResult.success();
+            });
+            PlayerMetaSyncService.pushIfChanged(player);async.removeFirst().run();
+            when(player.hasPermission("tfmc.flag")).thenReturn(true);PlayerMetaSyncService.pushIfChanged(player);async.removeFirst().run();
+            String current=site[0];assertTrue(current.contains("\"tfmc.flag\":true"));
+            site[0]=lateWrite[0];
+            now[0]+=180_000L;PlayerMetaSyncService.pushIfChanged(player);async.removeFirst().run();assertEquals(current,site[0]);
+            now[0]=1_000L+PlayerMetaSyncService.REPAIR_WINDOW_MS-1;PlayerMetaSyncService.pushIfChanged(player);async.removeFirst().run();
+            site[0]=lateWrite[0];
+            now[0]=1_000L+PlayerMetaSyncService.REPAIR_WINDOW_MS;PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());async.removeFirst().run();assertEquals(current,site[0]);
+            PlayerMetaSyncService.pushIfChanged(player);assertTrue(async.isEmpty());
+            // A new failure restarts the window.
+            timeOut[0]=true;PlayerMetaSyncService.pushForPlayer(player);async.removeFirst().run();
+            now[0]+=PlayerMetaSyncService.REPAIR_WINDOW_MS-1;PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());async.removeFirst().run();
+            PlayerMetaSyncService.pushIfChanged(player);assertEquals(1,async.size());async.removeFirst().run();
+            now[0]+=1;PlayerMetaSyncService.pushIfChanged(player);async.removeFirst().run();PlayerMetaSyncService.pushIfChanged(player);assertTrue(async.isEmpty());
+        } finally {PlayerMetaSyncService.clock=original;}
+    }
     @Test void syncEscapesControlCharactersAndUsesRootLocale() {
         Locale.setDefault(Locale.forLanguageTag("tr-TR"));Cache.realmId="MAIN";
         var meta=new EntitlementResolver.ResolvedMeta(0,false,false,1,1,1,0,List.of("ITEM"),false,Map.of("a\n\u0001",true));