WIP: новая схема сообщений и push (не проверено)

2026-04-19 20:41:58 +03:00
parent f0b560ec06
commit cc59bd18ee
27 changed files with 1668 additions and 94 deletions
@@ -59,14 +59,20 @@ import server.logic.ws_protocol.JSON.handlers.connections.entyties.Net_GetUserCo
 import server.logic.ws_protocol.JSON.handlers.connections.entyties.Net_AddCloseFriend_Request;
 import server.logic.ws_protocol.JSON.handlers.connections.entyties.Net_ListContacts_Request;
 import server.logic.ws_protocol.JSON.messages.Net_AckIncomingMessage_Handler;
+import server.logic.ws_protocol.JSON.messages.Net_AckSessionDelivery_Handler;
 import server.logic.ws_protocol.JSON.messages.Net_CallInviteBroadcast_Handler;
 import server.logic.ws_protocol.JSON.messages.Net_CallSignalToSession_Handler;
+import server.logic.ws_protocol.JSON.messages.Net_ReceiveIncomingMessage_Handler;
 import server.logic.ws_protocol.JSON.messages.Net_SendDirectMessage_Handler;
+import server.logic.ws_protocol.JSON.messages.Net_SendMessagePair_Handler;
 import server.logic.ws_protocol.JSON.messages.Net_UpsertPushToken_Handler;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_AckSessionDelivery_Request;
 import server.logic.ws_protocol.JSON.messages.entyties.Net_AckIncomingMessage_Request;
 import server.logic.ws_protocol.JSON.messages.entyties.Net_CallInviteBroadcast_Request;
 import server.logic.ws_protocol.JSON.messages.entyties.Net_CallSignalToSession_Request;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_ReceiveIncomingMessage_Request;
 import server.logic.ws_protocol.JSON.messages.entyties.Net_SendDirectMessage_Request;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_SendMessagePair_Request;
 import server.logic.ws_protocol.JSON.messages.entyties.Net_UpsertPushToken_Request;

 // --- NEW: Ping ---
@@ -124,7 +130,10 @@ public final class JsonHandlerRegistry {
            // --- direct messages / push ---
            Map.entry("UpsertPushToken", new Net_UpsertPushToken_Handler()),
            Map.entry("SendDirectMessage", new Net_SendDirectMessage_Handler()),
+            Map.entry("SendMessagePair", new Net_SendMessagePair_Handler()),
+            Map.entry("ReceiveIncomingMessage", new Net_ReceiveIncomingMessage_Handler()),
            Map.entry("AckIncomingMessage", new Net_AckIncomingMessage_Handler()),
+            Map.entry("AckSessionDelivery", new Net_AckSessionDelivery_Handler()),
            Map.entry("CallInviteBroadcast", new Net_CallInviteBroadcast_Handler()),
            Map.entry("CallSignalToSession", new Net_CallSignalToSession_Handler()),

@@ -172,7 +181,10 @@ public final class JsonHandlerRegistry {
            // --- direct messages / push ---
            Map.entry("UpsertPushToken", Net_UpsertPushToken_Request.class),
            Map.entry("SendDirectMessage", Net_SendDirectMessage_Request.class),
+            Map.entry("SendMessagePair", Net_SendMessagePair_Request.class),
+            Map.entry("ReceiveIncomingMessage", Net_ReceiveIncomingMessage_Request.class),
            Map.entry("AckIncomingMessage", Net_AckIncomingMessage_Request.class),
+            Map.entry("AckSessionDelivery", Net_AckSessionDelivery_Request.class),
            Map.entry("CallInviteBroadcast", Net_CallInviteBroadcast_Request.class),
            Map.entry("CallSignalToSession", Net_CallSignalToSession_Request.class),

@@ -10,6 +10,7 @@ import server.logic.ws_protocol.JSON.entyties.Net_Response;
 import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
 import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_CreateAuthSession_Request;
 import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_CreateAuthSession_Response;
+import server.logic.ws_protocol.JSON.messages.SignedMessagesRealtime;
 import server.logic.ws_protocol.JSON.utils.AuthKeyUtils;
 import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
 import server.logic.ws_protocol.WireCodes;
@@ -378,6 +379,7 @@ public class Net_CreateAuthSession__Handler implements JsonMessageHandler {
        ctx.setAuthenticationStatus(ConnectionContext.AUTH_STATUS_USER);

        ActiveConnectionsRegistry.getInstance().register(ctx);
+        SignedMessagesRealtime.dispatchPendingForSession(ctx);

        // --- формируем ответ ---
        Net_CreateAuthSession_Response resp = new Net_CreateAuthSession_Response();
@@ -10,6 +10,7 @@ import server.logic.ws_protocol.JSON.entyties.Net_Response;
 import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
 import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_SessionLogin_Request;
 import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_SessionLogin_Response;
+import server.logic.ws_protocol.JSON.messages.SignedMessagesRealtime;
 import server.logic.ws_protocol.JSON.utils.AuthKeyUtils;
 import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
 import server.logic.ws_protocol.WireCodes;
@@ -261,6 +262,7 @@ public class Net_SessionLogin_Handler implements JsonMessageHandler {
        ctx.setAuthenticationStatus(ConnectionContext.AUTH_STATUS_USER);

        ActiveConnectionsRegistry.getInstance().register(ctx);
+        SignedMessagesRealtime.dispatchPendingForSession(ctx);

        // ответ
        Net_SessionLogin_Response resp = new Net_SessionLogin_Response();
@@ -0,0 +1,34 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import server.logic.ws_protocol.JSON.ConnectionContext;
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_AckSessionDelivery_Request;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_AckSessionDelivery_Response;
+import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
+import server.logic.ws_protocol.WireCodes;
+import shine.db.dao.SignedMessagesV2DAO;
+
+public class Net_AckSessionDelivery_Handler implements JsonMessageHandler {
+    @Override
+    public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) throws Exception {
+        Net_AckSessionDelivery_Request req = (Net_AckSessionDelivery_Request) baseRequest;
+        if (ctx == null || !ctx.isAuthenticatedUser()) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.UNVERIFIED, "NOT_AUTHENTICATED", "Требуется авторизация");
+        }
+        if (req.getMessageKey() == null || req.getMessageKey().isBlank()) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_FIELDS", "messageKey обязателен");
+        }
+
+        String messageKey = req.getMessageKey().trim();
+        SignedMessagesV2DAO.getInstance().markDelivered(messageKey, ctx.getSessionId(), System.currentTimeMillis());
+
+        Net_AckSessionDelivery_Response resp = new Net_AckSessionDelivery_Response();
+        resp.setOp(req.getOp());
+        resp.setRequestId(req.getRequestId());
+        resp.setStatus(WireCodes.Status.OK);
+        resp.setMessageKey(messageKey);
+        return resp;
+    }
+}
@@ -0,0 +1,63 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import server.logic.ws_protocol.JSON.ConnectionContext;
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_ReceiveIncomingMessage_Request;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_ReceiveIncomingMessage_Response;
+import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
+import server.logic.ws_protocol.WireCodes;
+import shine.db.entities.SignedMessageV2Entry;
+
+public class Net_ReceiveIncomingMessage_Handler implements JsonMessageHandler {
+    @Override
+    public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) throws Exception {
+        Net_ReceiveIncomingMessage_Request req = (Net_ReceiveIncomingMessage_Request) baseRequest;
+        if (isBlank(req.getIncomingBlobB64())) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_FIELDS", "incomingBlobB64 обязателен");
+        }
+
+        final SignedMessageBlock incoming;
+        try {
+            incoming = SignedMessagesCore.parseFromB64(req.getIncomingBlobB64());
+        } catch (IllegalArgumentException ex) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, ex.getMessage(), "Некорректный формат входящего блока");
+        }
+        if (!incoming.isIncomingType()) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_MESSAGE_TYPE", "API принимает только входящие типы 1/3");
+        }
+
+        try {
+            SignedMessagesCore.verifyUsersAndSignature(incoming);
+        } catch (IllegalArgumentException ex) {
+            String code = ex.getMessage();
+            int status = "USER_NOT_FOUND".equals(code) ? 404 : WireCodes.Status.UNVERIFIED;
+            return NetExceptionResponseFactory.error(req, status, code, "Сообщение не прошло проверку");
+        }
+
+        final SignedMessageV2Entry entry;
+        try {
+            entry = SignedMessagesCore.toEntry(incoming, "ReceiveIncomingMessage", null);
+        } catch (IllegalArgumentException ex) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, ex.getMessage(), "Некорректный payload подтверждения");
+        }
+
+        SignedMessagesCore.saveIfAbsent(entry);
+        SignedMessagesRealtime.DeliveryCounters counters = SignedMessagesRealtime.deliverToTargetSessions(entry, null);
+
+        Net_ReceiveIncomingMessage_Response resp = new Net_ReceiveIncomingMessage_Response();
+        resp.setOp(req.getOp());
+        resp.setRequestId(req.getRequestId());
+        resp.setStatus(WireCodes.Status.OK);
+        resp.setMessageKey(entry.getMessageKey());
+        resp.setBaseKey(entry.getBaseKey());
+        resp.setDeliveredWsSessions(counters.wsDelivered);
+        resp.setDeliveredWebPushSessions(counters.pushDelivered);
+        return resp;
+    }
+
+    private boolean isBlank(String s) {
+        return s == null || s.isBlank();
+    }
+}
@@ -0,0 +1,81 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import server.logic.ws_protocol.JSON.ConnectionContext;
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_SendMessagePair_Request;
+import server.logic.ws_protocol.JSON.messages.entyties.Net_SendMessagePair_Response;
+import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
+import server.logic.ws_protocol.WireCodes;
+import shine.db.entities.SignedMessageV2Entry;
+
+public class Net_SendMessagePair_Handler implements JsonMessageHandler {
+    @Override
+    public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) throws Exception {
+        Net_SendMessagePair_Request req = (Net_SendMessagePair_Request) baseRequest;
+        if (ctx == null || !ctx.isAuthenticatedUser()) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.UNVERIFIED, "NOT_AUTHENTICATED", "Требуется авторизация");
+        }
+        if (isBlank(req.getIncomingBlobB64()) || isBlank(req.getOutgoingBlobB64())) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_FIELDS", "incomingBlobB64/outgoingBlobB64 обязательны");
+        }
+
+        final SignedMessageBlock incoming;
+        final SignedMessageBlock outgoing;
+        try {
+            incoming = SignedMessagesCore.parseFromB64(req.getIncomingBlobB64());
+            outgoing = SignedMessagesCore.parseFromB64(req.getOutgoingBlobB64());
+            SignedMessagesCore.validatePair(incoming, outgoing);
+        } catch (IllegalArgumentException ex) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, ex.getMessage(), "Некорректный формат пары сообщений");
+        }
+
+        if (!incoming.fromLogin.equalsIgnoreCase(ctx.getLogin())) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.UNVERIFIED, "SENDER_MISMATCH", "fromLogin должен совпадать с авторизованной сессией");
+        }
+
+        try {
+            SignedMessagesCore.verifyUsersAndSignature(incoming);
+            SignedMessagesCore.verifyUsersAndSignature(outgoing);
+        } catch (IllegalArgumentException ex) {
+            String code = ex.getMessage();
+            int status = "USER_NOT_FOUND".equals(code) ? 404 : WireCodes.Status.UNVERIFIED;
+            return NetExceptionResponseFactory.error(req, status, code, "Сообщение не прошло проверку");
+        }
+
+        SignedMessageV2Entry incomingEntry;
+        SignedMessageV2Entry outgoingEntry;
+        try {
+            incomingEntry = SignedMessagesCore.toEntry(incoming, "SendMessagePair", ctx.getSessionId());
+            outgoingEntry = SignedMessagesCore.toEntry(outgoing, "SendMessagePair", ctx.getSessionId());
+        } catch (IllegalArgumentException ex) {
+            return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, ex.getMessage(), "Некорректный payload подтверждения");
+        }
+
+        SignedMessagesCore.saveIfAbsent(incomingEntry);
+        SignedMessagesCore.saveIfAbsent(outgoingEntry);
+
+        SignedMessagesRealtime.DeliveryCounters inCounters =
+                SignedMessagesRealtime.deliverToTargetSessions(incomingEntry, null);
+
+        String excludeSessionId = outgoingEntry.getTargetLogin().equalsIgnoreCase(ctx.getLogin()) ? ctx.getSessionId() : null;
+        SignedMessagesRealtime.DeliveryCounters outCounters =
+                SignedMessagesRealtime.deliverToTargetSessions(outgoingEntry, excludeSessionId);
+
+        Net_SendMessagePair_Response resp = new Net_SendMessagePair_Response();
+        resp.setOp(req.getOp());
+        resp.setRequestId(req.getRequestId());
+        resp.setStatus(WireCodes.Status.OK);
+        resp.setBaseKey(incomingEntry.getBaseKey());
+        resp.setIncomingKey(incomingEntry.getMessageKey());
+        resp.setOutgoingKey(outgoingEntry.getMessageKey());
+        resp.setDeliveredWsSessions(inCounters.wsDelivered + outCounters.wsDelivered);
+        resp.setDeliveredWebPushSessions(inCounters.pushDelivered + outCounters.pushDelivered);
+        return resp;
+    }
+
+    private boolean isBlank(String s) {
+        return s == null || s.isBlank();
+    }
+}
@@ -0,0 +1,57 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+
+final class ReadReceiptPayload {
+    final String refToLogin;
+    final String refFromLogin;
+    final long refTimeMs;
+    final long refNonce;
+    final int refType;
+
+    private ReadReceiptPayload(String refToLogin, String refFromLogin, long refTimeMs, long refNonce, int refType) {
+        this.refToLogin = refToLogin;
+        this.refFromLogin = refFromLogin;
+        this.refTimeMs = refTimeMs;
+        this.refNonce = refNonce;
+        this.refType = refType;
+    }
+
+    static ReadReceiptPayload parse(byte[] payload) {
+        if (payload == null || payload.length < 1 + 1 + 8 + 4 + 2) {
+            throw new IllegalArgumentException("BAD_RECEIPT_PAYLOAD_LEN");
+        }
+        ByteBuffer bb = ByteBuffer.wrap(payload).order(ByteOrder.BIG_ENDIAN);
+        String refTo = readAscii(bb, 1, 60, "BAD_RECEIPT_TO_LOGIN");
+        String refFrom = readAscii(bb, 1, 60, "BAD_RECEIPT_FROM_LOGIN");
+        long refTimeMs = bb.getLong();
+        if (refTimeMs < 0) throw new IllegalArgumentException("BAD_RECEIPT_TIME");
+        long refNonce = Integer.toUnsignedLong(bb.getInt());
+        int refType = Short.toUnsignedInt(bb.getShort());
+        if (refType < SignedMessageBlock.TYPE_INCOMING_TEXT || refType > SignedMessageBlock.TYPE_READ_OUTGOING_COPY) {
+            throw new IllegalArgumentException("BAD_RECEIPT_REF_TYPE");
+        }
+        if (bb.hasRemaining()) {
+            throw new IllegalArgumentException("BAD_RECEIPT_PAYLOAD_LEN");
+        }
+        return new ReadReceiptPayload(refTo, refFrom, refTimeMs, refNonce, refType);
+    }
+
+    String refBaseKey() {
+        return SignedMessageKeys.baseKey(refToLogin, refFromLogin, refTimeMs, refNonce);
+    }
+
+    private static String readAscii(ByteBuffer bb, int minLen, int maxLen, String code) {
+        if (!bb.hasRemaining()) throw new IllegalArgumentException(code);
+        int len = Byte.toUnsignedInt(bb.get());
+        if (len < minLen || len > maxLen || bb.remaining() < len) throw new IllegalArgumentException(code);
+        byte[] bytes = new byte[len];
+        bb.get(bytes);
+        for (byte b : bytes) {
+            if (b < 0x20 || b > 0x7E) throw new IllegalArgumentException(code);
+        }
+        return new String(bytes, StandardCharsets.US_ASCII);
+    }
+}
@@ -0,0 +1,118 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+final class SignedMessageBlock {
+    static final byte[] PREFIX = "SHiNE_dm2".getBytes(StandardCharsets.US_ASCII);
+    static final int TYPE_INCOMING_TEXT = 1;
+    static final int TYPE_OUTGOING_COPY = 2;
+    static final int TYPE_READ_INCOMING = 3;
+    static final int TYPE_READ_OUTGOING_COPY = 4;
+
+    final String toLogin;
+    final String fromLogin;
+    final long timeMs;
+    final long nonce;
+    final int messageType;
+    final byte[] payloadBytes;
+    final byte[] signedBody;
+    final byte[] signature64;
+    final byte[] rawPacket;
+
+    private SignedMessageBlock(
+            String toLogin,
+            String fromLogin,
+            long timeMs,
+            long nonce,
+            int messageType,
+            byte[] payloadBytes,
+            byte[] signedBody,
+            byte[] signature64,
+            byte[] rawPacket
+    ) {
+        this.toLogin = toLogin;
+        this.fromLogin = fromLogin;
+        this.timeMs = timeMs;
+        this.nonce = nonce;
+        this.messageType = messageType;
+        this.payloadBytes = payloadBytes;
+        this.signedBody = signedBody;
+        this.signature64 = signature64;
+        this.rawPacket = rawPacket;
+    }
+
+    static SignedMessageBlock parse(byte[] raw, int maxPayloadBytes) {
+        if (raw == null || raw.length < PREFIX.length + 1 + 1 + 8 + 4 + 2 + 2 + 64) {
+            throw new IllegalArgumentException("BAD_LEN");
+        }
+        if (raw.length > 8192) {
+            throw new IllegalArgumentException("PAYLOAD_TOO_LARGE");
+        }
+
+        ByteBuffer bb = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN);
+        byte[] prefix = new byte[PREFIX.length];
+        bb.get(prefix);
+        if (!Arrays.equals(prefix, PREFIX)) {
+            throw new IllegalArgumentException("BAD_PREFIX");
+        }
+
+        String toLogin = readAscii(bb, 1, 60, "BAD_TO_LOGIN");
+        String fromLogin = readAscii(bb, 1, 60, "BAD_FROM_LOGIN");
+
+        long timeMs = bb.getLong();
+        if (timeMs < 0) throw new IllegalArgumentException("BAD_TIME");
+
+        long nonce = Integer.toUnsignedLong(bb.getInt());
+        int messageType = Short.toUnsignedInt(bb.getShort());
+        if (messageType < TYPE_INCOMING_TEXT || messageType > TYPE_READ_OUTGOING_COPY) {
+            throw new IllegalArgumentException("BAD_MESSAGE_TYPE");
+        }
+
+        int payloadLen = Short.toUnsignedInt(bb.getShort());
+        if (payloadLen < 1 || payloadLen > maxPayloadBytes) {
+            throw new IllegalArgumentException("BAD_MESSAGE_LEN");
+        }
+        if (bb.remaining() != payloadLen + 64) {
+            throw new IllegalArgumentException("BAD_LEN");
+        }
+
+        byte[] payload = new byte[payloadLen];
+        bb.get(payload);
+        byte[] signature64 = new byte[64];
+        bb.get(signature64);
+        byte[] signedBody = Arrays.copyOf(raw, raw.length - 64);
+
+        return new SignedMessageBlock(
+                toLogin, fromLogin, timeMs, nonce, messageType, payload, signedBody, signature64, raw
+        );
+    }
+
+    boolean isIncomingType() {
+        return messageType == TYPE_INCOMING_TEXT || messageType == TYPE_READ_INCOMING;
+    }
+
+    boolean isOutgoingCopyType() {
+        return messageType == TYPE_OUTGOING_COPY || messageType == TYPE_READ_OUTGOING_COPY;
+    }
+
+    String targetLogin() {
+        return isIncomingType() ? toLogin : fromLogin;
+    }
+
+    private static String readAscii(ByteBuffer bb, int minLen, int maxLen, String code) {
+        if (!bb.hasRemaining()) throw new IllegalArgumentException(code);
+        int len = Byte.toUnsignedInt(bb.get());
+        if (len < minLen || len > maxLen || bb.remaining() < len) {
+            throw new IllegalArgumentException(code);
+        }
+        byte[] bytes = new byte[len];
+        bb.get(bytes);
+        for (byte b : bytes) {
+            if (b < 0x20 || b > 0x7E) throw new IllegalArgumentException(code);
+        }
+        return new String(bytes, StandardCharsets.US_ASCII);
+    }
+}
@@ -0,0 +1,13 @@
+package server.logic.ws_protocol.JSON.messages;
+
+final class SignedMessageKeys {
+    private SignedMessageKeys() {}
+
+    static String baseKey(String toLogin, String fromLogin, long timeMs, long nonce) {
+        return fromLogin + "|" + toLogin + "|" + timeMs + "|" + nonce;
+    }
+
+    static String messageKey(String toLogin, String fromLogin, long timeMs, long nonce, int messageType) {
+        return baseKey(toLogin, fromLogin, timeMs, nonce) + "|" + messageType;
+    }
+}
@@ -0,0 +1,89 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import shine.db.dao.SignedMessagesV2DAO;
+import shine.db.dao.SolanaUsersDAO;
+import shine.db.entities.SignedMessageV2Entry;
+import shine.db.entities.SolanaUserEntry;
+import utils.crypto.Ed25519Util;
+
+import java.util.Base64;
+
+final class SignedMessagesCore {
+    private static final int MAX_PAYLOAD_BYTES = 4096;
+
+    private SignedMessagesCore() {}
+
+    static SignedMessageBlock parseFromB64(String blobB64) {
+        try {
+            byte[] raw = Base64.getDecoder().decode(blobB64.trim());
+            return SignedMessageBlock.parse(raw, MAX_PAYLOAD_BYTES);
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("BAD_BLOCK_FORMAT");
+        }
+    }
+
+    static void verifyUsersAndSignature(SignedMessageBlock block) throws Exception {
+        SolanaUserEntry from = SolanaUsersDAO.getInstance().getByLogin(block.fromLogin);
+        SolanaUserEntry to = SolanaUsersDAO.getInstance().getByLogin(block.toLogin);
+        if (from == null || to == null) {
+            throw new IllegalArgumentException("USER_NOT_FOUND");
+        }
+        byte[] pubKey32 = Ed25519Util.keyFromBase64(from.getDeviceKey());
+        if (!Ed25519Util.verify(block.signedBody, block.signature64, pubKey32)) {
+            throw new IllegalArgumentException("BAD_SIGNATURE");
+        }
+    }
+
+    static void validatePair(SignedMessageBlock incoming, SignedMessageBlock outgoing) {
+        if (incoming.messageType % 2 == 0) throw new IllegalArgumentException("BAD_PAIR_TYPES");
+        if (outgoing.messageType != incoming.messageType + 1) throw new IllegalArgumentException("BAD_PAIR_TYPES");
+        if (!incoming.toLogin.equalsIgnoreCase(outgoing.toLogin)) throw new IllegalArgumentException("BAD_PAIR_KEYS");
+        if (!incoming.fromLogin.equalsIgnoreCase(outgoing.fromLogin)) throw new IllegalArgumentException("BAD_PAIR_KEYS");
+        if (incoming.timeMs != outgoing.timeMs) throw new IllegalArgumentException("BAD_PAIR_KEYS");
+        if (incoming.nonce != outgoing.nonce) throw new IllegalArgumentException("BAD_PAIR_KEYS");
+
+        if (incoming.messageType == SignedMessageBlock.TYPE_READ_INCOMING) {
+            ReadReceiptPayload inRef = ReadReceiptPayload.parse(incoming.payloadBytes);
+            ReadReceiptPayload outRef = ReadReceiptPayload.parse(outgoing.payloadBytes);
+            if (!inRef.refToLogin.equalsIgnoreCase(outRef.refToLogin)
+                    || !inRef.refFromLogin.equalsIgnoreCase(outRef.refFromLogin)
+                    || inRef.refTimeMs != outRef.refTimeMs
+                    || inRef.refNonce != outRef.refNonce
+                    || inRef.refType != outRef.refType) {
+                throw new IllegalArgumentException("BAD_RECEIPT_REF");
+            }
+        }
+    }
+
+    static SignedMessageV2Entry toEntry(SignedMessageBlock block, String sourceApi, String originSessionId) {
+        String baseKey = SignedMessageKeys.baseKey(block.toLogin, block.fromLogin, block.timeMs, block.nonce);
+        String messageKey = SignedMessageKeys.messageKey(block.toLogin, block.fromLogin, block.timeMs, block.nonce, block.messageType);
+
+        SignedMessageV2Entry entry = new SignedMessageV2Entry();
+        entry.setMessageKey(messageKey);
+        entry.setBaseKey(baseKey);
+        entry.setTargetLogin(block.targetLogin());
+        entry.setFromLogin(block.fromLogin);
+        entry.setToLogin(block.toLogin);
+        entry.setTimeMs(block.timeMs);
+        entry.setNonce(block.nonce);
+        entry.setMessageType(block.messageType);
+        entry.setRawBlock(block.rawPacket);
+        entry.setCreatedAtMs(System.currentTimeMillis());
+        entry.setSourceApi(sourceApi);
+        entry.setOriginSessionId(originSessionId);
+
+        if (block.messageType == SignedMessageBlock.TYPE_READ_INCOMING
+                || block.messageType == SignedMessageBlock.TYPE_READ_OUTGOING_COPY) {
+            ReadReceiptPayload ref = ReadReceiptPayload.parse(block.payloadBytes);
+            entry.setReceiptRefBaseKey(ref.refBaseKey());
+            entry.setReceiptRefType(ref.refType);
+        }
+
+        return entry;
+    }
+
+    static void saveIfAbsent(SignedMessageV2Entry entry) throws Exception {
+        SignedMessagesV2DAO.getInstance().insertIfAbsent(entry);
+    }
+}
@@ -0,0 +1,136 @@
+package server.logic.ws_protocol.JSON.messages;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import server.logic.ws_protocol.JSON.ActiveConnectionsRegistry;
+import server.logic.ws_protocol.JSON.ConnectionContext;
+import server.logic.ws_protocol.JSON.push.WebPushSender;
+import server.logic.ws_protocol.JSON.push.WsEventSender;
+import shine.db.dao.ActiveSessionsDAO;
+import shine.db.dao.SignedMessagesV2DAO;
+import shine.db.entities.ActiveSessionEntry;
+import shine.db.entities.SignedMessageV2Entry;
+
+import java.util.Base64;
+import java.util.List;
+
+public final class SignedMessagesRealtime {
+    private static final Logger log = LoggerFactory.getLogger(SignedMessagesRealtime.class);
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final int LOGIN_BACKLOG_LIMIT = 500;
+
+    private SignedMessagesRealtime() {}
+
+    static DeliveryCounters deliverToTargetSessions(
+            SignedMessageV2Entry message,
+            String excludeSessionId
+    ) throws Exception {
+        DeliveryCounters counters = new DeliveryCounters();
+        List<ActiveSessionEntry> sessions = ActiveSessionsDAO.getInstance().getByLogin(message.getTargetLogin());
+        long now = System.currentTimeMillis();
+        for (ActiveSessionEntry s : sessions) {
+            String sessionId = s.getSessionId();
+            if (excludeSessionId != null && excludeSessionId.equals(sessionId)) {
+                continue;
+            }
+            SignedMessagesV2DAO.getInstance().ensureDeliveryRow(message.getMessageKey(), sessionId, now);
+            boolean deliveredOnline = sendEventToSessionIfOnline(sessionId, message, false);
+            if (deliveredOnline) {
+                counters.wsDelivered++;
+                continue;
+            }
+            if (message.getMessageType() == SignedMessageBlock.TYPE_INCOMING_TEXT) {
+                boolean pushed = pushNewMessageNotification(s, message);
+                if (pushed) counters.pushDelivered++;
+            }
+        }
+        return counters;
+    }
+
+    public static void dispatchPendingForSession(ConnectionContext ctx) {
+        if (ctx == null || !ctx.isAuthenticatedUser()) return;
+        String login = ctx.getLogin();
+        String sessionId = ctx.getSessionId();
+        if (isBlank(login) || isBlank(sessionId)) return;
+
+        try {
+            List<SignedMessageV2Entry> pending = SignedMessagesV2DAO.getInstance()
+                    .listPendingForSession(login, sessionId, LOGIN_BACKLOG_LIMIT);
+            for (SignedMessageV2Entry e : pending) {
+                sendEventToSessionIfOnline(sessionId, e, true);
+            }
+        } catch (Exception e) {
+            log.warn("Failed to dispatch pending messages for sessionId={}", sessionId, e);
+        }
+    }
+
+    private static boolean sendEventToSessionIfOnline(String sessionId, SignedMessageV2Entry message, boolean backlog) {
+        ConnectionContext targetCtx = ActiveConnectionsRegistry.getInstance().getBySessionId(sessionId);
+        if (targetCtx == null) return false;
+
+        String blobB64 = Base64.getEncoder().encodeToString(message.getRawBlock());
+        ObjectNode payload = MAPPER.createObjectNode();
+        payload.put("messageKey", message.getMessageKey());
+        payload.put("baseKey", message.getBaseKey());
+        payload.put("fromLogin", message.getFromLogin());
+        payload.put("toLogin", message.getToLogin());
+        payload.put("targetLogin", message.getTargetLogin());
+        payload.put("messageType", message.getMessageType());
+        payload.put("timeMs", message.getTimeMs());
+        payload.put("nonce", message.getNonce());
+        payload.put("blobB64", blobB64);
+        payload.put("backlog", backlog);
+        if (message.getReceiptRefBaseKey() != null) {
+            payload.put("receiptRefBaseKey", message.getReceiptRefBaseKey());
+        }
+        if (message.getReceiptRefType() != null) {
+            payload.put("receiptRefType", message.getReceiptRefType());
+        }
+        return WsEventSender.sendEvent(targetCtx, "SignedMessageArrived", message.getMessageKey(), payload);
+    }
+
+    private static boolean pushNewMessageNotification(ActiveSessionEntry session, SignedMessageV2Entry message) {
+        try {
+            if (session == null) return false;
+            if (isBlank(session.getPushEndpoint()) || isBlank(session.getPushP256dhKey()) || isBlank(session.getPushAuthKey())) {
+                return false;
+            }
+            String text = "Вам пришло сообщение от " + message.getFromLogin() + ". Откройте для прочтения.";
+            String payload = "{\"kind\":\"new_message\",\"fromLogin\":\"" + jsonEscape(message.getFromLogin()) + "\",\"text\":\"" + jsonEscape(text) + "\"}";
+            return WebPushSender.sendBase64Payload(
+                    session.getPushEndpoint(),
+                    session.getPushP256dhKey(),
+                    session.getPushAuthKey(),
+                    payload
+            );
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    private static String jsonEscape(String s) {
+        if (s == null) return "";
+        StringBuilder out = new StringBuilder();
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            if (c == '\\') out.append("\\\\");
+            else if (c == '"') out.append("\\\"");
+            else if (c == '\n') out.append("\\n");
+            else if (c == '\r') out.append("\\r");
+            else if (c == '\t') out.append("\\t");
+            else out.append(c);
+        }
+        return out.toString();
+    }
+
+    private static boolean isBlank(String s) {
+        return s == null || s.isBlank();
+    }
+
+    static final class DeliveryCounters {
+        int wsDelivered;
+        int pushDelivered;
+    }
+}
@@ -0,0 +1,10 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+
+public class Net_AckSessionDelivery_Request extends Net_Request {
+    private String messageKey;
+
+    public String getMessageKey() { return messageKey; }
+    public void setMessageKey(String messageKey) { this.messageKey = messageKey; }
+}
@@ -0,0 +1,10 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+
+public class Net_AckSessionDelivery_Response extends Net_Response {
+    private String messageKey;
+
+    public String getMessageKey() { return messageKey; }
+    public void setMessageKey(String messageKey) { this.messageKey = messageKey; }
+}
@@ -0,0 +1,10 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+
+public class Net_ReceiveIncomingMessage_Request extends Net_Request {
+    private String incomingBlobB64;
+
+    public String getIncomingBlobB64() { return incomingBlobB64; }
+    public void setIncomingBlobB64(String incomingBlobB64) { this.incomingBlobB64 = incomingBlobB64; }
+}
@@ -0,0 +1,19 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+
+public class Net_ReceiveIncomingMessage_Response extends Net_Response {
+    private String messageKey;
+    private String baseKey;
+    private int deliveredWsSessions;
+    private int deliveredWebPushSessions;
+
+    public String getMessageKey() { return messageKey; }
+    public void setMessageKey(String messageKey) { this.messageKey = messageKey; }
+    public String getBaseKey() { return baseKey; }
+    public void setBaseKey(String baseKey) { this.baseKey = baseKey; }
+    public int getDeliveredWsSessions() { return deliveredWsSessions; }
+    public void setDeliveredWsSessions(int deliveredWsSessions) { this.deliveredWsSessions = deliveredWsSessions; }
+    public int getDeliveredWebPushSessions() { return deliveredWebPushSessions; }
+    public void setDeliveredWebPushSessions(int deliveredWebPushSessions) { this.deliveredWebPushSessions = deliveredWebPushSessions; }
+}
@@ -0,0 +1,13 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+
+public class Net_SendMessagePair_Request extends Net_Request {
+    private String incomingBlobB64;
+    private String outgoingBlobB64;
+
+    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; }
+}
@@ -0,0 +1,22 @@
+package server.logic.ws_protocol.JSON.messages.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+
+public class Net_SendMessagePair_Response extends Net_Response {
+    private String baseKey;
+    private String incomingKey;
+    private String outgoingKey;
+    private int deliveredWsSessions;
+    private int deliveredWebPushSessions;
+
+    public String getBaseKey() { return baseKey; }
+    public void setBaseKey(String baseKey) { this.baseKey = baseKey; }
+    public String getIncomingKey() { return incomingKey; }
+    public void setIncomingKey(String incomingKey) { this.incomingKey = incomingKey; }
+    public String getOutgoingKey() { return outgoingKey; }
+    public void setOutgoingKey(String outgoingKey) { this.outgoingKey = outgoingKey; }
+    public int getDeliveredWsSessions() { return deliveredWsSessions; }
+    public void setDeliveredWsSessions(int deliveredWsSessions) { this.deliveredWsSessions = deliveredWsSessions; }
+    public int getDeliveredWebPushSessions() { return deliveredWebPushSessions; }
+    public void setDeliveredWebPushSessions(int deliveredWebPushSessions) { this.deliveredWebPushSessions = deliveredWebPushSessions; }
+}