Сервер: локальный роутинг и relay DM по access servers

SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserAccessServersCurrentDAO.java

@@ -0,0 +1,66 @@
+package shine.db.dao;
+
+import shine.db.DbController;
+import shine.db.KeyEncodingUtil;
+import shine.db.entities.UserAccessServerRouteEntry;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * DAO локальной routing-проекции access servers пользователей.
+ */
+public final class UserAccessServersCurrentDAO {
+
+    private static volatile UserAccessServersCurrentDAO instance;
+    private final DbController db = DbController.getInstance();
+
+    private UserAccessServersCurrentDAO() {}
+
+    public static UserAccessServersCurrentDAO getInstance() {
+        if (instance == null) {
+            synchronized (UserAccessServersCurrentDAO.class) {
+                if (instance == null) instance = new UserAccessServersCurrentDAO();
+            }
+        }
+        return instance;
+    }
+
+    public List<UserAccessServerRouteEntry> listByUserLogin(String userLogin) throws SQLException {
+        try (Connection c = db.getConnection()) {
+            return listByUserLogin(c, userLogin);
+        }
+    }
+
+    public List<UserAccessServerRouteEntry> listByUserLogin(Connection c, String userLogin) throws SQLException {
+        String sql = """
+            SELECT user_login, server_login, server_url, server_client_key
+            FROM user_access_servers_current
+            WHERE LOWER(user_login) = LOWER(?)
+            ORDER BY server_login
+            """;
+        List<UserAccessServerRouteEntry> result = new ArrayList<>();
+        try (PreparedStatement ps = c.prepareStatement(sql)) {
+            ps.setString(1, userLogin);
+            try (ResultSet rs = ps.executeQuery()) {
+                while (rs.next()) {
+                    result.add(mapRow(rs));
+                }
+            }
+        }
+        return result;
+    }
+
+    private UserAccessServerRouteEntry mapRow(ResultSet rs) throws SQLException {
+        UserAccessServerRouteEntry entry = new UserAccessServerRouteEntry();
+        entry.setUserLogin(rs.getString("user_login"));
+        entry.setServerLogin(rs.getString("server_login"));
+        entry.setServerUrl(rs.getString("server_url"));
+        entry.setServerClientKey(KeyEncodingUtil.normalizeKeyToBase64_32(rs.getString("server_client_key")));
+        return entry;
+    }
+}

SHiNE-server/shine-server-db/src/main/java/shine/db/entities/UserAccessServerRouteEntry.java

@@ -0,0 +1,27 @@
+package shine.db.entities;
+
+/**
+ * Локальная routing-проекция access server пользователя.
+ *
+ * Источник:
+ *  - user_access_servers_current
+ */
+public class UserAccessServerRouteEntry {
+
+    private String userLogin;
+    private String serverLogin;
+    private String serverUrl;
+    private String serverClientKey;
+
+    public String getUserLogin() { return userLogin; }
+    public void setUserLogin(String userLogin) { this.userLogin = userLogin; }
+
+    public String getServerLogin() { return serverLogin; }
+    public void setServerLogin(String serverLogin) { this.serverLogin = serverLogin; }
+
+    public String getServerUrl() { return serverUrl; }
+    public void setServerUrl(String serverUrl) { this.serverUrl = serverUrl; }
+
+    public String getServerClientKey() { return serverClientKey; }
+    public void setServerClientKey(String serverClientKey) { this.serverClientKey = serverClientKey; }
+}

SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_ReceiveIncomingMessage_Handler.java

@@ -53,6 +53,11 @@ public class Net_ReceiveIncomingMessage_Handler implements JsonMessageHandler {
         SignedMessagesRealtime.DeliveryCounters counters = new SignedMessagesRealtime.DeliveryCounters();
         if (status.applied()) {
             counters = SignedMessagesRealtime.deliverToRelevantSessions(entry, incoming);
+            server.sync.DmFederationService.fanOutIncomingToRecipientAccessServers(
+                    incoming.toLogin,
+                    req.getIncomingBlobB64().trim(),
+                    req.getSourceServerLogin()
+            );
         }
         if (status == SignedMessagesDAO.ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE) {
             bounceConversationDeleteIfKnown(incoming.fromLogin, incoming.toLogin);

SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_ReceiveIncomingMessage_Request.java

@@ -4,7 +4,10 @@ import server.logic.ws_protocol.JSON.entyties.Net_Request;
 
 public class Net_ReceiveIncomingMessage_Request extends Net_Request {
     private String incomingBlobB64;
+    private String sourceServerLogin;
 
     public String getIncomingBlobB64() { return incomingBlobB64; }
     public void setIncomingBlobB64(String incomingBlobB64) { this.incomingBlobB64 = incomingBlobB64; }
+    public String getSourceServerLogin() { return sourceServerLogin; }
+    public void setSourceServerLogin(String sourceServerLogin) { this.sourceServerLogin = sourceServerLogin; }
 }

SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_SendMessagePair_Request.java

@@ -5,9 +5,12 @@ import server.logic.ws_protocol.JSON.entyties.Net_Request;
 public class Net_SendMessagePair_Request extends Net_Request {
     private String incomingBlobB64;
     private String outgoingBlobB64;
+    private String sourceServerLogin;
 
     public String getIncomingBlobB64() { return incomingBlobB64; }
     public void setIncomingBlobB64(String incomingBlobB64) { this.incomingBlobB64 = incomingBlobB64; }
     public String getOutgoingBlobB64() { return outgoingBlobB64; }
     public void setOutgoingBlobB64(String outgoingBlobB64) { this.outgoingBlobB64 = outgoingBlobB64; }
+    public String getSourceServerLogin() { return sourceServerLogin; }
+    public void setSourceServerLogin(String sourceServerLogin) { this.sourceServerLogin = sourceServerLogin; }
 }

SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java

@@ -2,7 +2,8 @@ package server.sync;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import server.logic.ws_protocol.JSON.handlers.auth.SolanaUserPdaImportService;
+import shine.db.dao.UserAccessServersCurrentDAO;
+import shine.db.entities.UserAccessServerRouteEntry;
 import utils.config.AppConfig;
 
 import java.util.LinkedHashMap;
@@ -18,26 +19,48 @@ public final class DmFederationService {
 
     public static void fanOutPair(String fromLogin, String toLogin, String incomingBlobB64, String outgoingBlobB64) {
         try {
-            Map<String, SolanaUserPdaImportService.ParsedServerRoute> senderRoutes =
+            Map<String, UserAccessServerRouteEntry> senderRoutes =
-                    routesByLogin(SolanaUserPdaImportService.fetchAccessServerRoutesByLogin(fromLogin));
+                    routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(fromLogin));
-            Map<String, SolanaUserPdaImportService.ParsedServerRoute> recipientRoutes =
+            Map<String, UserAccessServerRouteEntry> recipientRoutes =
-                    routesByLogin(SolanaUserPdaImportService.fetchAccessServerRoutesByLogin(toLogin));
+                    routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin));
 
             String ownServerLogin = ownServerLogin();
-            for (SolanaUserPdaImportService.ParsedServerRoute route : senderRoutes.values()) {
+            for (UserAccessServerRouteEntry route : senderRoutes.values()) {
                 if (isOwnServer(route, ownServerLogin)) continue;
-                REMOTE.sendMessagePair(route.serverAddress(), incomingBlobB64, outgoingBlobB64);
+                REMOTE.sendMessagePair(route.getServerUrl(), incomingBlobB64, outgoingBlobB64, ownServerLogin);
             }
-            for (SolanaUserPdaImportService.ParsedServerRoute route : recipientRoutes.values()) {
+            for (UserAccessServerRouteEntry route : recipientRoutes.values()) {
                 if (isOwnServer(route, ownServerLogin)) continue;
-                if (senderRoutes.containsKey(route.login())) continue;
+                if (senderRoutes.containsKey(normalize(route.getServerLogin()))) continue;
-                REMOTE.receiveIncomingMessage(route.serverAddress(), incomingBlobB64);
+                REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin);
             }
         } catch (Exception e) {
             log.warn("DM federation pair fan-out failed: from={} to={}", fromLogin, toLogin, e);
         }
     }
 
+    public static void fanOutIncomingToRecipientAccessServers(
+            String toLogin,
+            String incomingBlobB64,
+            String sourceServerLogin
+    ) {
+        try {
+            Map<String, UserAccessServerRouteEntry> recipientRoutes =
+                    routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin));
+            String ownServerLogin = ownServerLogin();
+            String normalizedSource = normalize(sourceServerLogin);
+            for (UserAccessServerRouteEntry route : recipientRoutes.values()) {
+                String routeLogin = normalize(route.getServerLogin());
+                if (routeLogin == null) continue;
+                if (ownServerLogin != null && ownServerLogin.equalsIgnoreCase(routeLogin)) continue;
+                if (normalizedSource != null && normalizedSource.equalsIgnoreCase(routeLogin)) continue;
+                REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin);
+            }
+        } catch (Exception e) {
+            log.warn("DM federation incoming relay failed: to={}", toLogin, e);
+        }
+    }
+
     public static void fanOutDeleteMessage(String fromLogin, String toLogin, int messageType, String blobB64) {
         fanOutSingleDelete(fromLogin, toLogin, messageType, blobB64, true);
     }
@@ -48,17 +71,17 @@ public final class DmFederationService {
 
     private static void fanOutSingleDelete(String fromLogin, String toLogin, int messageType, String blobB64, boolean oneMessageDelete) {
         try {
-            Map<String, SolanaUserPdaImportService.ParsedServerRoute> routes = routesByLogin(
+            Map<String, UserAccessServerRouteEntry> routes = routesByLogin(
-                    SolanaUserPdaImportService.fetchAccessServerRoutesByLogin(fromLogin),
+                    UserAccessServersCurrentDAO.getInstance().listByUserLogin(fromLogin),
-                    SolanaUserPdaImportService.fetchAccessServerRoutesByLogin(toLogin)
+                    UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin)
             );
             String ownServerLogin = ownServerLogin();
-            for (SolanaUserPdaImportService.ParsedServerRoute route : routes.values()) {
+            for (UserAccessServerRouteEntry route : routes.values()) {
                 if (isOwnServer(route, ownServerLogin)) continue;
                 if (oneMessageDelete) {
-                    REMOTE.deleteMessage(route.serverAddress(), blobB64);
+                    REMOTE.deleteMessage(route.getServerUrl(), blobB64);
                 } else {
-                    REMOTE.deleteConversation(route.serverAddress(), blobB64);
+                    REMOTE.deleteConversation(route.getServerUrl(), blobB64);
                 }
             }
         } catch (Exception e) {
@@ -68,15 +91,15 @@ public final class DmFederationService {
     }
 
     @SafeVarargs
-    private static Map<String, SolanaUserPdaImportService.ParsedServerRoute> routesByLogin(
+    private static Map<String, UserAccessServerRouteEntry> routesByLogin(
-            List<SolanaUserPdaImportService.ParsedServerRoute>... routeLists
+            List<UserAccessServerRouteEntry>... routeLists
     ) {
-        Map<String, SolanaUserPdaImportService.ParsedServerRoute> out = new LinkedHashMap<>();
+        Map<String, UserAccessServerRouteEntry> out = new LinkedHashMap<>();
-        for (List<SolanaUserPdaImportService.ParsedServerRoute> routeList : routeLists) {
+        for (List<UserAccessServerRouteEntry> routeList : routeLists) {
-            for (SolanaUserPdaImportService.ParsedServerRoute route : routeList) {
+            for (UserAccessServerRouteEntry route : routeList) {
                 if (route == null) continue;
-                String login = normalize(route.login());
+                String login = normalize(route.getServerLogin());
-                String address = route.serverAddress() == null ? "" : route.serverAddress().trim();
+                String address = route.getServerUrl() == null ? "" : route.getServerUrl().trim();
                 if (login == null || address.isBlank()) continue;
                 out.putIfAbsent(login, route);
             }
@@ -84,8 +107,11 @@ public final class DmFederationService {
         return out;
     }
 
-    private static boolean isOwnServer(SolanaUserPdaImportService.ParsedServerRoute route, String ownServerLogin) {
+    private static boolean isOwnServer(UserAccessServerRouteEntry route, String ownServerLogin) {
-        return ownServerLogin != null && ownServerLogin.equalsIgnoreCase(route.login());
+        return ownServerLogin != null
+                && route != null
+                && route.getServerLogin() != null
+                && ownServerLogin.equalsIgnoreCase(route.getServerLogin());
     }
 
     private static String ownServerLogin() {

SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmSyncClient.java

@@ -21,33 +21,35 @@ public final class RemoteDmSyncClient {
             .connectTimeout(Duration.ofSeconds(6))
             .build();
 
-    public void sendMessagePair(String serverAddressRaw, String incomingBlobB64, String outgoingBlobB64) throws Exception {
+    public void sendMessagePair(String serverAddressRaw, String incomingBlobB64, String outgoingBlobB64, String sourceServerLogin) throws Exception {
         String incomingJson = MAPPER.writeValueAsString(incomingBlobB64);
         String outgoingJson = MAPPER.writeValueAsString(outgoingBlobB64);
+        String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin);
         JsonNode response = send(serverAddressRaw, """
             {
               "op":"ReceiveOutcomingMessage",
               "requestId":%s,
               "payload":{
                 "incomingBlobB64":%s,
-                "outgoingBlobB64":%s
+                "outgoingBlobB64":%s%s
               }
             }
-            """.formatted("%s", incomingJson, outgoingJson));
+            """.formatted("%s", incomingJson, outgoingJson, sourceServerLoginJson));
         ensureOk("ReceiveOutcomingMessage", response);
     }
 
-    public void receiveIncomingMessage(String serverAddressRaw, String incomingBlobB64) throws Exception {
+    public void receiveIncomingMessage(String serverAddressRaw, String incomingBlobB64, String sourceServerLogin) throws Exception {
         String incomingJson = MAPPER.writeValueAsString(incomingBlobB64);
+        String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin);
         JsonNode response = send(serverAddressRaw, """
             {
               "op":"ReceiveIncomingMessage",
               "requestId":%s,
               "payload":{
-                "incomingBlobB64":%s
+                "incomingBlobB64":%s%s
               }
             }
-            """.formatted("%s", incomingJson));
+            """.formatted("%s", incomingJson, sourceServerLoginJson));
         ensureOk("ReceiveIncomingMessage", response);
     }
 
@@ -107,6 +109,13 @@ public final class RemoteDmSyncClient {
         return MAPPER.readTree(responseJson);
     }
 
+    private String toOptionalJsonField(String fieldName, String value) throws Exception {
+        if (value == null || value.isBlank()) {
+            return "";
+        }
+        return ",\n                \"" + fieldName + "\":" + MAPPER.writeValueAsString(value.trim());
+    }
+
     private void ensureOk(String op, JsonNode response) {
         int status = response.path("status").asInt(500);
         if (status >= 200 && status < 300) return;

VERSION.properties

@@ -1,2 +1,2 @@
 client.version=1.2.355
-server.version=1.2.339
+server.version=1.2.340