Add atomic user mutation transaction extension

AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataManager.java

@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.Objects;
 import java.util.Map.Entry;
+import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -36,6 +37,7 @@
 import com.bencodez.advancedcore.core.user.runtime.SharedUserDataRuntime;
 import com.bencodez.simpleapi.debug.DebugLevel;
 import com.bencodez.simpleapi.array.ArrayUtils;
+import com.bencodez.simpleapi.sql.data.DataValue;
 
 import lombok.Getter;
 
@@ -325,6 +327,20 @@ public final boolean hasSharedRuntime() {
 		return runtime != null && !runtime.isClosed();
 	}
 
+	/**
+	 * Cache-safe transaction entry for callers adding their own durable SQL
+	 * record to a user mutation. Initial values are row prerequisites and may
+	 * commit before queued cache work; put operation-specific mutation only in
+	 * the callback. Blocking work must run off the game thread.
+	 */
+	public final <T> T withAtomicUserTransaction(UUID uuid, UserStorage storage,
+			Map<String, DataValue> initialValues,
+			SqlUserStorage.TransactionWork<T> work) {
+		SharedUserDataRuntime runtime = sharedRuntime;
+		if (runtime == null) throw new IllegalStateException("Shared user runtime is unavailable");
+		return runtime.transaction(uuid, storage, initialValues, work);
+	}
+
 	/**
 	 * Replace the active shared backend on this manager's worker. The runtime
 	 * flushes and retires the old cache generation before publishing the new
@@ -969,6 +985,11 @@ private void reportDeferredStorageFailure(Throwable failure) {
 		}
 	}
 
+	/** Retain a cache reconciliation failure after its SQL transaction committed. */
+	public final void recordSharedStorageFailure(Throwable failure) {
+		reportDeferredStorageFailure(Objects.requireNonNull(failure, "failure"));
+	}
+
 	/** Last asynchronous cache-cleanup failure, retained for diagnosis and recovery. */
 	public Throwable getLastDeferredStorageFailure() { return lastDeferredStorageFailure.get(); }
 

AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/userstorage/mysql/MySQL.java

@@ -125,16 +125,20 @@ public void loadData() {
 		uuids.clear();
 		uuids.addAll(getUuidsQuery());
 
-		names.clear();
-		names.addAll(getNamesQuery());
+		synchronized (names) {
+			names.clear();
+			names.addAll(getNamesQuery());
+		}
 	}
 
 	public void clearCacheBasic() {
 		clearCaches();
 		uuids.clear();
 		uuids.addAll(getUuidsQuery());
-		names.clear();
-		names.addAll(getNamesQuery());
+		synchronized (names) {
+			names.clear();
+			names.addAll(getNamesQuery());
+		}
 	}
 
 	// -------------------------
@@ -360,6 +364,12 @@ public void forEachUser(java.util.function.BiConsumer<UUID, ArrayList<Column>> p
 	// Keep existing methods (getUuids / getUUID / etc.)
 	// -------------------------
 
+	/** Publish a committed UUID; refresh names only when PlayerName may have changed. */
+	public void recordCommittedUser(UUID uuid, boolean nameMayHaveChanged) {
+		uuids.add(uuid.toString());
+		if (nameMayHaveChanged) synchronized (names) { names.clear(); }
+	}
+
 	public Set<String> getUuids() {
 		if (uuids == null || uuids.isEmpty()) {
 			uuids.clear();
@@ -385,13 +395,12 @@ public ArrayList<String> getUuidsQuery() {
 		return out;
 	}
 
-	public Set<String> getNames() {
-		if (names == null || names.isEmpty()) {
-			names.clear();
-			names.addAll(getNamesQuery());
-		}
-		return names;
-	}
+	public Set<String> getNames() {
+		synchronized (names) {
+			if (names.isEmpty()) names.addAll(getNamesQuery());
+			return new java.util.HashSet<>(names);
+		}
+	}
 
 	public ArrayList<String> getNamesQuery() {
 		ArrayList<String> out = new ArrayList<>();
@@ -576,7 +585,7 @@ public void deletePlayer(String uuid) {
 		}
 
 		uuids.remove(uuid);
-		names.remove(UuidLookup.getInstance().getCachedName(uuid));
+		synchronized (names) { names.remove(UuidLookup.getInstance().getCachedName(uuid)); }
 		clearCacheBasic();
 	}
 
@@ -816,12 +825,11 @@ public void insertQuery(String index, List<Column> cols) {
 				playerName = col.getValue().toString();
 			}
 		}
-		if (playerName == null || playerName.isEmpty()) {
-			names.add(UuidLookup.getInstance().getPlayerName(
-					plugin.getUserManager().getUser(java.util.UUID.fromString(index), false), index, false));
-		} else {
-			names.add(playerName);
-		}
+		String committedName = playerName == null || playerName.isEmpty()
+				? UuidLookup.getInstance().getPlayerName(
+						plugin.getUserManager().getUser(java.util.UUID.fromString(index), false), index, false)
+				: playerName;
+		synchronized (names) { names.add(committedName); }
 
 		uuids.add(index);
 		plugin.devDebug("Inserting " + index + " into database");

AdvancedCore/src/main/java/com/bencodez/advancedcore/bukkit/user/runtime/BukkitUserCacheOwner.java

@@ -145,6 +145,11 @@ private void bind(UserDataCache cache, UUID uuid) {
 
     @Override public boolean isCached(UUID uuid) { return manager.isCached(uuid); }
 
+    @Override public boolean hasPendingChanges(UUID uuid) {
+        UserDataCache cache = manager.getUserDataCache().get(uuid);
+        return cache != null && cache.hasChangesToProcess();
+    }
+
     @Override public DataValue getIfPresent(UUID uuid, String key) {
         UserDataCache cache = manager.getUserDataCache().get(uuid);
         if (cache == null) return null;
@@ -253,6 +258,10 @@ private record CachePopulation(UUID uuid, UserDataCache cache, long version) imp
 		for (UUID uuid : Set.copyOf(pendingNotifications.keySet())) dispatchNotifications(uuid);
 	}
 
+    @Override public void reportCommittedFailure(UUID uuid, Throwable failure) {
+        manager.recordSharedStorageFailure(failure);
+    }
+
 	@Override public void discardAllNotifications() { pendingNotifications.clear(); }
 
     @Override public Set<UUID> cachedUsers() { return new HashSet<>(manager.getUserDataCache().keySet()); }

AdvancedCore/src/main/java/com/bencodez/advancedcore/bukkit/user/storage/BukkitSqlUserBackend.java

@@ -7,13 +7,18 @@
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.sql.DriverManager;
+import java.sql.SQLException;
 
 import com.bencodez.advancedcore.AdvancedCorePlugin;
 import com.bencodez.advancedcore.api.user.UserStorage;
 import com.bencodez.advancedcore.api.user.userstorage.mysql.MySQL;
 import com.bencodez.advancedcore.api.user.userstorage.sql.UserTable;
 import com.bencodez.advancedcore.core.user.storage.SqlUserStorage;
+import com.bencodez.advancedcore.core.user.storage.sql.SqlBackendLogger;
 import com.bencodez.advancedcore.core.user.storage.sql.SqlUserBackend;
+import com.bencodez.advancedcore.core.user.storage.sql.SqlUserBackendFactory;
+import com.bencodez.advancedcore.core.user.storage.sql.SqlUserSchema;
 import com.bencodez.simpleapi.sql.Column;
 import com.bencodez.simpleapi.sql.data.DataValue;
 import com.bencodez.simpleapi.sql.data.DataValueString;
@@ -74,6 +79,12 @@ public SqlUserStorage user(UUID uuid) {
             @Override public void writeValues(UserStorage storage, HashMap<String, DataValue> values) {
                 BukkitSqlUserBackend.this.writeValues(storage, uuid, values);
             }
+            @Override public <T> T transaction(UserStorage storage, TransactionWork<T> work) {
+                return BukkitSqlUserBackend.this.transaction(storage, uuid, java.util.Map.of(), work);
+            }
+            @Override public <T> T transaction(UserStorage storage, java.util.Map<String, DataValue> initialValues, TransactionWork<T> work) {
+                return BukkitSqlUserBackend.this.transaction(storage, uuid, initialValues, work);
+            }
         };
     }
 
@@ -148,13 +159,88 @@ private void writeValues(UserStorage storage, UUID uuid, HashMap<String, DataVal
         Objects.requireNonNull(values, "values");
         requireOpen();
         requireStorage(storage);
-        ArrayList<Column> columns = new ArrayList<>();
+        java.util.Map<String, DataValue> updates = new HashMap<>();
         values.forEach((key, value) -> {
-            if (!"uuid".equalsIgnoreCase(key) && value != null) columns.add(new Column(key, value));
+            if (!"uuid".equalsIgnoreCase(key) && value != null) updates.put(key, value);
+        });
+        if (updates.isEmpty()) return;
+        // Shared cache flushes must surface SQL failures. The legacy update
+        // methods log and swallow them, which could acknowledge a later caller
+        // transaction while its prerequisite queued values were never stored.
+        withSqlUser(storage, uuid, updates, user -> {
+            user.writeValues(storage, new HashMap<>(updates));
+            return null;
+        });
+        if (storage == UserStorage.MYSQL) mysql().recordCommittedUser(uuid, containsPlayerName(updates));
+    }
+
+    private <T> T transaction(UserStorage storage, UUID uuid, java.util.Map<String, DataValue> initialValues, SqlUserStorage.TransactionWork<T> work) {
+        requireOpen();
+        requireStorage(storage);
+        Objects.requireNonNull(work, "work");
+        return withSqlUser(storage, uuid, initialValues, user -> {
+            if (storage != UserStorage.MYSQL) return user.transaction(storage, initialValues, work);
+            boolean[] nameTouched = {false};
+            T result = user.transaction(storage, initialValues, scope -> {
+                nameTouched[0] = scope.createdUserRow() && containsPlayerName(initialValues);
+                return work.run(new SqlUserStorage.TransactionScope() {
+                    @Override public boolean createdUserRow() { return scope.createdUserRow(); }
+                    @Override public java.sql.Connection connection() { return scope.connection(); }
+                    @Override public List<Column> readRow() throws SQLException { return scope.readRow(); }
+                    @Override public void writeValues(java.util.Map<String, DataValue> values) throws SQLException {
+                        scope.writeValues(values);
+                        if (containsPlayerName(values)) nameTouched[0] = true;
+                    }
+                });
+            });
+            mysql().recordCommittedUser(uuid, nameTouched[0]);
+            return result;
         });
-        if (columns.isEmpty()) return;
-        if (storage == UserStorage.MYSQL) mysql().update(uuid.toString(), columns, false);
-        else synchronized (sqliteOperations) { table().update(primary(uuid), columns); }
+    }
+
+    private static boolean containsPlayerName(java.util.Map<String, DataValue> values) {
+        return values.keySet().stream().anyMatch("PlayerName"::equalsIgnoreCase);
+    }
+
+    private <T> T withSqlUser(UserStorage storage, UUID uuid, java.util.Map<String, DataValue> updates,
+            java.util.function.Function<SqlUserStorage, T> work) {
+        SqlUserSchema registered = SqlUserSchema.fromKeys(plugin.getUserManager().getDataManager().getKeys());
+        SqlUserSchema.Builder builder = SqlUserSchema.builder();
+        for (SqlUserSchema.ColumnDefinition column : registered.columns()) {
+            if (!"uuid".equalsIgnoreCase(column.name())) builder.column(column.name(), column.sqlType(), column.dataType());
+        }
+        ArrayList<Column> dynamic = new ArrayList<>();
+        for (java.util.Map.Entry<String, DataValue> entry : updates.entrySet()) {
+            if (registered.contains(entry.getKey())) continue;
+            com.bencodez.simpleapi.sql.DataType type = entry.getValue().getType();
+            builder.column(entry.getKey(), type == com.bencodez.simpleapi.sql.DataType.STRING ? "TEXT" :
+                    type == com.bencodez.simpleapi.sql.DataType.INTEGER ? "INTEGER" : "BOOLEAN", type);
+            dynamic.add(new Column(entry.getKey(), type));
+        }
+        SqlUserSchema schema = builder.build();
+        SqlBackendLogger logger = new SqlBackendLogger() {
+            @Override public void info(String message) { plugin.getLogger().info(message); }
+            @Override public void warn(String message, Throwable error) {
+                plugin.getLogger().warning(message + (error == null ? "" : ": " + error.getMessage()));
+            }
+        };
+        if (storage == UserStorage.MYSQL) {
+            for (Column column : dynamic) mysql().checkColumn(column.getName(), column.getDataType());
+            var manager = mysql().getMysql().getConnectionManager();
+            return work.apply(SqlUserBackendFactory.existingUser(storage, uuid, mysql().getTableName(), schema,
+                    manager::getConnection, manager.getDbType(), logger));
+        }
+        synchronized (sqliteOperations) {
+            for (Column column : dynamic) table().checkColumn(column);
+            // The legacy SQLite provider retains one shared connection. Obtain
+            // its actual database URL, then open a separate transaction-owned
+            // connection so AdvancedCore can close it after commit/rollback.
+            String url;
+            try { url = table().getSqLite().getSQLConnection().getMetaData().getURL(); }
+            catch (SQLException failure) { throw new IllegalStateException("Failed to locate Bukkit SQLite user database", failure); }
+            return work.apply(SqlUserBackendFactory.existingUser(storage, uuid, table().getName(), schema,
+                    () -> DriverManager.getConnection(url), null, logger));
+        }
     }
 
     private Column primary(UUID uuid) { return new Column("uuid", new DataValueString(uuid.toString())); }

AdvancedCore/src/main/java/com/bencodez/advancedcore/core/user/runtime/SharedUserDataRuntime.java

@@ -2,6 +2,7 @@
 
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
@@ -16,7 +17,9 @@
 import java.util.function.Supplier;
 
 import com.bencodez.advancedcore.api.user.UserDataFetchMode;
+import com.bencodez.advancedcore.api.user.UserStorage;
 import com.bencodez.advancedcore.core.user.storage.SqlUserDataAccess;
+import com.bencodez.advancedcore.core.user.storage.SqlUserStorage;
 import com.bencodez.advancedcore.core.user.storage.sql.SqlUserBackend;
 import com.bencodez.simpleapi.sql.Column;
 import com.bencodez.simpleapi.sql.data.DataValue;
@@ -119,6 +122,82 @@ public void flush(UUID uuid) {
 
     private void flushInternal(UUID uuid) { cacheOwner.flush(uuid, backend.storageType(), backend.user(uuid)); }
 
+    /**
+     * Extend the active SQL backend's user transaction with caller-owned SQL.
+     * Pending cache changes are flushed first. Only after commit is the old
+     * cache generation retired, so a later load observes committed values and
+     * a rolled-back callback cannot publish speculative values.
+     */
+    public <T> T transaction(UUID uuid, SqlUserStorage.TransactionWork<T> work) {
+        return transactionInternal(uuid, null, Map.of(), work);
+    }
+
+    /**
+     * Require the selected physical store. Initial values are prerequisites
+     * for creating a row with required columns, not part of the caller's atomic
+     * mutation. When a cache already exists, they can commit before its queued
+     * changes are flushed; callers must write operation-specific values only
+     * inside the transaction callback.
+     */
+    public <T> T transaction(UUID uuid, UserStorage expectedStorage, Map<String, DataValue> initialValues,
+            SqlUserStorage.TransactionWork<T> work) {
+        Objects.requireNonNull(expectedStorage, "expectedStorage");
+        return transactionInternal(uuid, expectedStorage, initialValues, work);
+    }
+
+    private <T> T transactionInternal(UUID uuid, UserStorage expectedStorage, Map<String, DataValue> initialValues,
+            SqlUserStorage.TransactionWork<T> work) {
+        Objects.requireNonNull(uuid, "uuid");
+        Objects.requireNonNull(initialValues, "initialValues");
+        Objects.requireNonNull(work, "work");
+        cacheOwner.requireBlockingAllowed();
+        try {
+            return userExclusiveAccess(uuid, () -> {
+                if (expectedStorage != null && backend.storageType() != expectedStorage) {
+                    throw new IllegalStateException("Cannot access " + expectedStorage
+                            + " user storage while the shared runtime owns " + backend.storageType());
+                }
+                cacheOwner.beginRemoval(uuid);
+                try {
+                    if (cacheOwner.hasPendingChanges(uuid) && !initialValues.isEmpty()) {
+                        // Existing queued changes may need a required column on
+                        // first write. Establish only row prerequisites before
+                        // their ordinary, independently durable cache flush.
+                        backend.user(uuid).transaction(backend.storageType(), initialValues, scope -> null);
+                    }
+                    flushInternal(uuid);
+                } catch (RuntimeException | Error failure) {
+                    cacheOwner.cancelRemoval(uuid);
+                    throw failure;
+                }
+                // SQL has committed. Cache retirement can fail, but reporting a
+                // transaction failure here would invite a duplicate caller retry.
+                T result;
+                try {
+                    result = backend.user(uuid).transaction(backend.storageType(), initialValues, work);
+                } catch (RuntimeException | Error failure) {
+                    cacheOwner.cancelRemoval(uuid);
+                    throw failure;
+                }
+                try {
+                    cacheOwner.remove(uuid);
+                } catch (RuntimeException | Error failure) {
+                    try {
+                        cacheOwner.populate(uuid, SqlUserDataAccess.convert(readStorageRow(uuid)));
+                    } catch (RuntimeException | Error recoveryFailure) {
+                        failure.addSuppressed(recoveryFailure);
+                    } finally {
+                        try { cacheOwner.cancelRemoval(uuid); }
+                        catch (RuntimeException | Error recoveryFailure) { failure.addSuppressed(recoveryFailure); }
+                    }
+                    try { cacheOwner.reportCommittedFailure(uuid, failure); }
+                    catch (RuntimeException | Error reportingFailure) { failure.addSuppressed(reportingFailure); }
+                }
+                return result;
+            });
+        } finally { cacheOwner.dispatchNotifications(uuid); }
+    }
+
     public void flushAll() {
 		try { storageAccess(() -> {
             for (UUID uuid : Set.copyOf(cacheOwner.cachedUsers())) userAccess(uuid, () -> { flushInternal(uuid); return null; });