Add Redis MQTT and MySQL proxy transports for NeoForge

VotingPlugin/pom.xml

@@ -126,6 +126,19 @@
                         </filter>
                     </filters>
                     <relocations>
+                        <!-- Keep external JDBC driver-name strings unchanged in proxy consumers.
+                             Relocate the private Driver's JVM paths separately; ordinary relocation
+                             continues to rewrite its sibling classes and service providers. -->
+                        <relocation>
+                            <pattern>org.mariadb</pattern>
+                            <shadedPattern>${project.groupId}.votingplugin.neoforge.mariadb</shadedPattern>
+                            <excludes><exclude>org.mariadb.jdbc.Driver</exclude></excludes>
+                        </relocation>
+                        <relocation>
+                            <pattern>org/mariadb/jdbc/Driver</pattern>
+                            <shadedPattern>com/bencodez/votingplugin/neoforge/mariadb/jdbc/Driver</shadedPattern>
+                            <rawString>true</rawString>
+                        </relocation>
                         <relocation>
                             <pattern>com.tcoded.folialib</pattern>
                             <shadedPattern>
@@ -204,6 +217,13 @@
                             <minimizeJar>false</minimizeJar>
                             <createDependencyReducedPom>false</createDependencyReducedPom>
                             <filters>
+                                <filter>
+                                    <artifact>org.mariadb.jdbc:mariadb-java-client</artifact>
+                                    <excludes>
+                                        <!-- The NeoForge transport selects its private driver explicitly. -->
+                                        <exclude>META-INF/services/java.sql.Driver</exclude>
+                                    </excludes>
+                                </filter>
                                 <!-- Keep this PR buildable against both the current published
                                      AdvancedCore SNAPSHOT and its coordinated slimmer successor. -->
                                 <filter>
@@ -268,10 +288,38 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>3.8.1</version>
+                <executions>
+                    <execution>
+                        <id>neoforge-platform-gson-fixture</id>
+                        <phase>generate-test-resources</phase>
+                        <goals><goal>copy</goal></goals>
+                        <configuration>
+                            <artifactItems>
+                                <artifactItem>
+                                    <groupId>com.google.code.gson</groupId>
+                                    <artifactId>gson</artifactId>
+                                    <!-- Minecraft 1.21.1's supplied library; test input only. -->
+                                    <version>2.10.1</version>
+                                    <outputDirectory>${project.build.directory}/test-platform-libraries</outputDirectory>
+                                </artifactItem>
+                            </artifactItems>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
                 <version>3.5.4</version>
+                <configuration>
+                    <systemPropertyVariables>
+                        <votingplugin.neoforgeGsonJar>${project.build.directory}/test-platform-libraries/gson-2.10.1.jar</votingplugin.neoforgeGsonJar>
+                    </systemPropertyVariables>
+                </configuration>
                 <executions>
                     <execution>
                         <id>default-test</id>
@@ -358,6 +406,11 @@
         </repository>
     </repositories>
     <dependencies>
+        <dependency>
+            <groupId>org.mariadb.jdbc</groupId>
+            <artifactId>mariadb-java-client</artifactId>
+            <version>3.5.10</version>
+        </dependency>
         <dependency>
             <groupId>net.neoforged</groupId>
             <artifactId>neoforge</artifactId>

VotingPlugin/src/main/java/com/bencodez/votingplugin/neoforge/NeoForgeJsonStrings.java

@@ -0,0 +1,40 @@
+package com.bencodez.votingplugin.neoforge;
+
+import java.io.IOException;
+
+/** Strict JSON string lexicon for bounded inputs; callers enforce the object and value grammar. */
+final class NeoForgeJsonStrings {
+    private NeoForgeJsonStrings() { }
+
+    static void validate(String json) throws IOException {
+        // Gson 2.10.1's setLenient(false) still accepts raw controls and legacy string escapes.
+        // Validate the string lexicon before the streaming reader without depending on newer Gson APIs.
+        boolean quoted = false;
+        for (int i = 0; i < json.length(); i++) {
+            char c = json.charAt(i);
+            if (!quoted) {
+                if (c == '"') quoted = true;
+                continue;
+            }
+            if (c == '"') { quoted = false; continue; }
+            if (c < 0x20) throw invalid();
+            if (c != '\\') continue;
+            if (++i == json.length()) throw invalid();
+            switch (json.charAt(i)) {
+                case '"', '\\', '/', 'b', 'f', 'n', 'r', 't' -> { }
+                case 'u' -> {
+                    if (json.length() - i <= 4) throw invalid();
+                    for (int end = i + 4; i < end;) {
+                        char digit = json.charAt(++i);
+                        if (!(digit >= '0' && digit <= '9' || digit >= 'a' && digit <= 'f'
+                                || digit >= 'A' && digit <= 'F')) throw invalid();
+                    }
+                }
+                default -> throw invalid();
+            }
+        }
+        if (quoted) throw invalid();
+    }
+
+    private static IOException invalid() { return new IOException("Invalid JSON string"); }
+}

VotingPlugin/src/main/java/com/bencodez/votingplugin/neoforge/NeoForgeMysqlDataSource.java

@@ -0,0 +1,81 @@
+package com.bencodez.votingplugin.neoforge;
+
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.logging.Logger;
+
+import javax.sql.DataSource;
+
+import org.mariadb.jdbc.Configuration;
+import org.mariadb.jdbc.Driver;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+
+/** The bundled, privately relocated driver is selected directly, without global JDBC registration lookup. */
+final class NeoForgeMysqlDataSource {
+    private NeoForgeMysqlDataSource() { }
+
+    static Configuration configuration(NeoForgeProxyTransportConfiguration.Mysql settings) {
+        // Programmatic fields keep database names and opaque credentials out of URL option parsing.
+        return new Configuration.Builder().addHost(settings.host(), settings.port()).database(settings.database())
+                .user(settings.username()).password(settings.password()).connectTimeout(2000).socketTimeout(2000)
+                .sslMode(settings.ssl() ? "verify-full" : "disable")
+                .allowPublicKeyRetrieval(settings.publicKeyRetrieval()).allowLocalInfile(false)
+                .allowMultiQueries(false).dumpQueriesOnException(false).build();
+    }
+
+    static HikariDataSource open(NeoForgeProxyTransportConfiguration.Mysql settings) {
+        HikariConfig pool = new HikariConfig();
+        pool.setPoolName("VotingPlugin-NeoForge-MySQL");
+        pool.setDataSource(new Source(configuration(settings)));
+        pool.setMaximumPoolSize(settings.maxConnections());
+        pool.setMinimumIdle(0);
+        pool.setConnectionTimeout(2000);
+        pool.setValidationTimeout(1000);
+        pool.setInitializationFailTimeout(-1);
+        return new HikariDataSource(pool);
+    }
+
+    private static final class Source implements DataSource {
+        private final Configuration configuration;
+        Source(Configuration configuration) {
+            // Direct use still initializes Connector/J's global registration.
+            // Remove only this private driver class before admitting pool work,
+            // including startup/connection failures that never reach shutdown.
+            new Driver();
+            var registered = java.sql.DriverManager.getDrivers();
+            while (registered.hasMoreElements()) {
+                java.sql.Driver driver = registered.nextElement();
+                if (driver.getClass() == Driver.class) {
+                    try { java.sql.DriverManager.deregisterDriver(driver); }
+                    catch (SQLException failure) {
+                        throw new IllegalStateException("Unable to isolate transport driver");
+                    }
+                }
+            }
+            this.configuration = configuration;
+        }
+        @Override public Connection getConnection() throws SQLException { return Driver.connect(configuration); }
+        @Override public Connection getConnection(String user, String password) throws SQLException {
+            throw new SQLFeatureNotSupportedException("Use the configured transport credentials");
+        }
+        @Override public PrintWriter getLogWriter() { return null; }
+        @Override public void setLogWriter(PrintWriter writer) throws SQLException {
+            if (writer != null) throw new SQLFeatureNotSupportedException("Transport diagnostics are redacted");
+        }
+        @Override public int getLoginTimeout() { return 2; }
+        @Override public void setLoginTimeout(int seconds) throws SQLException {
+            if (seconds < 0) throw new SQLException("Invalid login timeout");
+            // The immutable driver configuration retains the stricter two-second connection bound.
+        }
+        @Override public Logger getParentLogger() { return Logger.getLogger("VotingPlugin-NeoForge-MySQL"); }
+        @Override public <T> T unwrap(Class<T> type) throws SQLException {
+            if (type.isInstance(this)) return type.cast(this);
+            throw new SQLException("Unsupported datasource wrapper");
+        }
+        @Override public boolean isWrapperFor(Class<?> type) { return type.isInstance(this); }
+    }
+}

VotingPlugin/src/main/java/com/bencodez/votingplugin/neoforge/NeoForgeMysqlQueue.java

@@ -0,0 +1,171 @@
+package com.bencodez.votingplugin.neoforge;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.nio.CharBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Logger;
+
+import javax.sql.DataSource;
+
+import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope;
+import com.bencodez.simpleapi.servercomm.codec.JsonEnvelopeCodec;
+
+/** Uses the existing SimpleAPI queue schema, retaining a row until application acceptance. */
+final class NeoForgeMysqlQueue implements NeoForgeProxyMysqlService.Queue {
+    static final int PAGE_SIZE = 50;
+    private static final int QUERY_SECONDS = 2;
+    private static final Logger LOGGER = Logger.getLogger(NeoForgeMysqlQueue.class.getName());
+    private static final String TABLE = "VotingPlugin_message_queue";
+    private final DataSource source;
+    private final AutoCloseable closeSource;
+    private final String server;
+    private final AtomicBoolean open = new AtomicBoolean(true);
+    private final AtomicBoolean invalidWarning = new AtomicBoolean();
+    private long cursor;
+
+    NeoForgeMysqlQueue(DataSource source, AutoCloseable closeSource, String server) throws SQLException {
+        this.source = Objects.requireNonNull(source, "source");
+        this.closeSource = Objects.requireNonNull(closeSource, "closeSource");
+        this.server = Objects.requireNonNull(server, "server");
+        try (Connection connection = source.getConnection(); var statement = connection.createStatement()) {
+            statement.setQueryTimeout(QUERY_SECONDS);
+            statement.execute("CREATE TABLE IF NOT EXISTS " + TABLE
+                    + " (id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, source VARCHAR(36) NOT NULL,"
+                    + " destination VARCHAR(36) NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,"
+                    + " payload LONGTEXT NOT NULL, PRIMARY KEY (id), INDEX idx_dest_id (destination, id)) ENGINE=InnoDB");
+        }
+    }
+
+    @Override public void poll(NeoForgeProxyMysqlService.Receiver receiver) throws SQLException {
+        if (!open.get()) return;
+        List<Row> rows = readPage();
+        // No pool connection remains checked out during storage work or acknowledgement publication.
+        for (Row row : rows) {
+            if (!open.get()) return;
+            if (row.payload == null) {
+                if (invalidWarning.compareAndSet(false, true)) LOGGER.warning("Retained invalid MySQL queue payload");
+                continue;
+            }
+            if (receiver.receive(row.source, row.destination, row.payload) && open.get()) remove(row);
+        }
+    }
+
+    private List<Row> readPage() throws SQLException {
+        ArrayList<Row> rows = new ArrayList<>(PAGE_SIZE);
+        try (Connection connection = source.getConnection()) {
+            String lock = "backend-channel-" + server;
+            if (!acquire(connection, lock)) return rows;
+            try {
+                // Bound the actual server response too: a large LONGTEXT never crosses the JDBC boundary.
+                try (PreparedStatement select = connection.prepareStatement("SELECT id, source, destination,"
+                        + " CASE WHEN OCTET_LENGTH(payload) <= " + NeoForgeProxyEnvelopeCodec.MAX_BYTES
+                        + " THEN payload ELSE NULL END AS payload FROM " + TABLE
+                        + " WHERE destination = ? AND id > ? ORDER BY id LIMIT " + PAGE_SIZE)) {
+                    select.setQueryTimeout(QUERY_SECONDS);
+                    select.setMaxRows(PAGE_SIZE);
+                    select.setString(1, server);
+                    select.setLong(2, cursor);
+                    try (var result = select.executeQuery()) {
+                        while (rows.size() < PAGE_SIZE && result.next()) {
+                            long id = result.getLong("id");
+                            if (id <= cursor) throw new SQLException("Invalid MySQL queue ordering");
+                            byte[] payload;
+                            try (Reader reader = result.getCharacterStream("payload")) {
+                                payload = boundedPayload(reader);
+                            } catch (IOException failure) {
+                                throw new SQLException("MySQL queue read failed");
+                            }
+                            rows.add(new Row(id, result.getString("source"), result.getString("destination"), payload));
+                            cursor = id;
+                        }
+                    }
+                }
+                // Revisit refused rows, while full pages can advance to other players rather than starve them.
+                if (rows.size() < PAGE_SIZE) cursor = 0;
+            } finally {
+                try (PreparedStatement release = connection.prepareStatement("SELECT RELEASE_LOCK(?)")) {
+                    release.setQueryTimeout(QUERY_SECONDS);
+                    release.setString(1, lock);
+                    try (var result = release.executeQuery()) {
+                        if (!result.next() || result.getInt(1) != 1) throw new SQLException("MySQL queue lock not released");
+                    }
+                } catch (SQLException failure) {
+                    // Returning a connection with a named lock to the pool can strand peer consumers.
+                    try { connection.abort(Runnable::run); }
+                    catch (SQLException cleanup) { failure.addSuppressed(cleanup); }
+                    throw failure;
+                }
+            }
+        }
+        return rows;
+    }
+
+    private static byte[] boundedPayload(Reader reader) throws IOException {
+        if (reader == null) return null;
+        StringBuilder text = new StringBuilder();
+        char[] buffer = new char[4096];
+        int count;
+        while ((count = reader.read(buffer)) != -1) {
+            if (text.length() + count > NeoForgeProxyEnvelopeCodec.MAX_BYTES) return null;
+            text.append(buffer, 0, count);
+        }
+        try {
+            var encoded = StandardCharsets.UTF_8.newEncoder().onMalformedInput(CodingErrorAction.REPORT)
+                    .onUnmappableCharacter(CodingErrorAction.REPORT).encode(CharBuffer.wrap(text));
+            if (encoded.remaining() > NeoForgeProxyEnvelopeCodec.MAX_BYTES) return null;
+            byte[] bytes = new byte[encoded.remaining()];
+            encoded.get(bytes);
+            return bytes;
+        } catch (CharacterCodingException invalid) { return null; }
+    }
+
+    private static boolean acquire(Connection connection, String lock) throws SQLException {
+        try (PreparedStatement statement = connection.prepareStatement("SELECT GET_LOCK(?, 0)")) {
+            statement.setQueryTimeout(QUERY_SECONDS);
+            statement.setString(1, lock);
+            try (var result = statement.executeQuery()) { return result.next() && result.getInt(1) == 1; }
+        }
+    }
+
+    private void remove(Row row) throws SQLException {
+        // Another consumer may have already removed it; a missing row is an idempotent outcome.
+        try (Connection connection = source.getConnection(); PreparedStatement remove = connection.prepareStatement(
+                "DELETE FROM " + TABLE + " WHERE id = ? AND source = 'proxy' AND destination = ?")) {
+            remove.setQueryTimeout(QUERY_SECONDS);
+            remove.setLong(1, row.id);
+            remove.setString(2, server);
+            remove.executeUpdate();
+        }
+    }
+
+    @Override public void publish(JsonEnvelope envelope) throws SQLException {
+        if (!open.get()) throw new SQLException("MySQL transport is closed");
+        String payload = JsonEnvelopeCodec.encode(envelope);
+        if (payload.getBytes(StandardCharsets.UTF_8).length > NeoForgeProxyEnvelopeCodec.MAX_BYTES) {
+            throw new SQLException("MySQL envelope exceeds transport limit");
+        }
+        try (Connection connection = source.getConnection(); PreparedStatement insert = connection.prepareStatement(
+                "INSERT INTO " + TABLE + " (source, destination, payload) VALUES (?, 'proxy', ?)")) {
+            insert.setQueryTimeout(QUERY_SECONDS);
+            insert.setString(1, server);
+            insert.setString(2, payload);
+            insert.executeUpdate();
+        }
+    }
+
+    @Override public void close() throws Exception {
+        if (open.compareAndSet(true, false)) closeSource.close();
+    }
+
+    private record Row(long id, String source, String destination, byte[] payload) { }
+}