Игнорирование отстуствия файлов APNs/FCM при разработке

.env

@@ -14,6 +14,10 @@ SDU_SERVERS=http://10.211.55.2:7777
 
 #Firebase Credentials
 FIREBASE_CREDENTIALS_PATH=serviceAccount.json
+#Apple APNS
+APNS_KEY_PATH=voip.p12
+APNS_P12_PASSWORD=rosetta1
+IOS_BUNDLE_ID=com.rosetta.dev
 
 #Каждые сколько дней будет очищаться буфер (максимальная дистанция синхронизации сообщений)
 BUFFER_CLEANUP_DAYS=7

.gitignore

@@ -34,6 +34,9 @@ target
 .settings
 .project
 .classpath
+serviceAccount.json
+build/*.p12
+*.p12
 
 build/.env
 build/.env*

pom.xml

@@ -46,6 +46,12 @@
             <artifactId>jakarta.persistence-api</artifactId>
             <version>3.1.0</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.eatthepath</groupId>
+            <artifactId>pushy</artifactId>
+            <version>0.15.4</version>
+        </dependency>
     </dependencies>
 
     <build>

serviceAccount.json

@@ -1,13 +0,0 @@
-{
-  "type": "service_account",
-  "project_id": "rosetta-messanger-dev",
-  "private_key_id": "69822675ff0a49a8ce0bd147dbab3ef432963485",
-  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC/iYhrgWfi/ecP\nIs7Y0u7MueE/leU0QZF3V6hIcs1Iq1/CSUYcszv+YXuKItfwi0pYtntPmnCtdKnv\nFgZZN/9I0/AjFPlggqEXBH223FjsAfoNTk8a0PXyF8oOG0u2pN34t20nxXFSYars\n/6yJMKOBdaJ+L51dlBeCj40UFm3zQSVpejRkspWmT5Qz7kWh2e8yj6ukPyXEfrrD\n8jmM6uQeKwzlmfjD8YbL7R2m+lE1hzkoFCo0PrqMQYIIpAMT9n9meCk/LrQMIUnA\n+zICsTffFwbFzmJ2A022po3ObL4dUYgqGrPs5DqjxwTF4mI0g2dkPvoxpUKkVzC7\nHzzOYTwRAgMBAAECggEAGIacDSa1rpdpfK2DLrjFHY+YXAPYehpX7fVaVeXmtcKa\nppkAXSYcQq9T2jUqUSGyOZgq7l9Yb0rhhFJd1L3QDBDLY9/zK3GpU/ZeF0oG1DhP\n6bVDJCwUBNXD/eNujKrMG5AimOBLlEtvpMemXa4jOa1eSxR+F4tM1AndUZr/pZFT\nIdPBrId3WemkcWH2b1R4RLVu8zqci1x+nWvL4Gt8O/gDpDzlmV2Disvpf300a5wU\ngIKcOI2H6vouOwL9ltMPTBXdpDIC64Q8zvK3973d3Fqx2Mt2mO4ZgylFpk2nmAm6\n3wknrkJDE+Z/H7XmdwpMhRmmyku3vuqsDwmy7BEGGwKBgQD03oOJ6eb9kG6JX830\nRH/Vhh3NlpVEzuHzRA8WznsX5WaeyMF9NWo4Lh9ifBxkRAQKftEC0CR5GZp3/895\nQuoMMbrrklWWkVRwCZWYQBquRXNHbTPj6ztvHaZZrw8HhemZWtOF89T5f8Kn4LcD\nFwR8kIqXeLxeTbfc2hQXZvP7vwKBgQDIPmhM2Td/Ni4Mbe3wGmOyS0RweK5qcKBF\n32N1rCr8smrD1irg+w9Qrmz2cX3udtVJDRLz4ctSmiEPs+UrzuyG6yTbdv7bh8AN\nsqHOdc4GRqGsB/jfXagVjRMSeObQk5fR03tiPLP15J+lpoEsjwUzw7L3Ioacaq62\nqSi87br8LwKBgQDu/uMJz4bJg5evcxeUSusuH5mlGE0WfIniIlJL4zoXR6qSXcUk\nDOdgb/vn5tTbM9tx1vbvNPH0VH4Ek2QPqbTANCWJWSk6LRxpwaEFmcOwxk5Or5IO\n6X/34suDCy6zHAu0xwZe3m7HGeCGc/iMBoI1heoPDyNjM5257AviD3UhBwKBgCWS\nmC17QI+NEfzhD5lSykwlFVVpP4jXUytpLBdjU7mQnLncULVgRlJkOCvRxchd4c1Q\nN7MtNeJs6zEwFxsuO3FhY8wOOunkQeQQFY5QynShAiruYANBZo2Mp/x6VQzj9MO5\nQ9h9/WJxIIeLg4dh2p8I5Ga8wrdMyTWa7frtPH2fAoGALUkccsu4Mcws55Xy1aHH\nNETI6CRQpaU4v6jNhuuY4g6jp5ScaqkpuuzHOYorpmdC3180YXzPn2pwKeYWa0BA\nTjR8z603bCpdRfpic1UvYi96jwAcNiJTAGGVwmyYopIFBdHphfGS3hWZlYfonVps\n+k87j3WGwOiUGqoKSGJi1cw=\n-----END PRIVATE KEY-----\n",
-  "client_email": "firebase-adminsdk-fbsvc@rosetta-messanger-dev.iam.gserviceaccount.com",
-  "client_id": "115421173243098464717",
-  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
-  "token_uri": "https://oauth2.googleapis.com/token",
-  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
-  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-fbsvc%40rosetta-messanger-dev.iam.gserviceaccount.com",
-  "universe_domain": "googleapis.com"
-}

src/main/java/im/rosetta/Boot.java

@@ -1,5 +1,6 @@
 package im.rosetta;
 
+import im.rosetta.calls.CallManager;
 import im.rosetta.client.ClientManager;
 import im.rosetta.client.OnlineManager;
 import im.rosetta.event.EventManager;
@@ -82,6 +83,7 @@ public class Boot {
     private OnlineManager onlineManager;
     private BufferCleanupService bufferCleanupService;
     private ForwardUnitService forwardUnitService;
+    private CallManager callManager;
 
     /**
      * Конструктор по умолчанию, использует порт 3000 для сервера
@@ -104,7 +106,8 @@ public class Boot {
             port, 
             30
         ), packetManager, this.serverAdapter);
-        this.clientManager = new ClientManager(server);
+        this.clientManager = new ClientManager(this.server);
+        this.callManager = new CallManager(this.clientManager);
         /**
          * Каждые сколько дней будет очищаться буфер (это влияет на синхронизацию сообщений, так 
          * как при синхронизации клиент запрашивает пакеты из буфера за последние 7 дней, если этот параметр будет меньше, 
@@ -113,7 +116,7 @@ public class Boot {
         int cleanupEveryDays = System.getenv("BUFFER_CLEANUP_DAYS") != null ? 
             Integer.parseInt(System.getenv("BUFFER_CLEANUP_DAYS")) : 7;
         this.bufferCleanupService = new BufferCleanupService(cleanupEveryDays, this.logger);
-        this.forwardUnitService = new ForwardUnitService(this.logger, this.clientManager);
+        this.forwardUnitService = new ForwardUnitService(this.logger, this.clientManager, this.callManager);
     }
 
     /**
@@ -228,8 +231,8 @@ public class Boot {
         this.packetManager.registerExecutor(22, new Executor22GroupBan());
         this.packetManager.registerExecutor(24, new Executor24DeviceResolve(this.clientManager, this.eventManager, this.packetManager));
         this.packetManager.registerExecutor(25, new Executor25Sync(this.packetManager));
-        this.packetManager.registerExecutor(26, new Executor26SignalPeer(this.clientManager, this.forwardUnitService));
-        this.packetManager.registerExecutor(27, new Executor27WebRTC(this.forwardUnitService));
+        this.packetManager.registerExecutor(26, new Executor26SignalPeer(this.clientManager, this.forwardUnitService, this.callManager));
+        this.packetManager.registerExecutor(27, new Executor27WebRTC(this.callManager));
         this.packetManager.registerExecutor(28, new Executor28IceServers(this.forwardUnitService));
     }
 

src/main/java/im/rosetta/Failures.java

@@ -27,7 +27,11 @@ public enum Failures implements BaseFailures {
     /**
      * Слишком много подписок на онлайн статусы
      */
-    TOO_MANY_ONLINE_SUBSCRIPTIONS(3010);
+    TOO_MANY_ONLINE_SUBSCRIPTIONS(3010),
+    /**
+     * Нет сессии звонка
+     */
+    NO_CALL_SESSION(3011);
 
     private final int code;
 

src/main/java/im/rosetta/calls/CallManager.java

@@ -0,0 +1,119 @@
+package im.rosetta.calls;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import im.rosetta.client.ClientManager;
+import im.rosetta.packet.Packet26SignalPeer;
+import im.rosetta.packet.runtime.NetworkSignalType;
+import io.g365sfu.Room;
+import io.orprotocol.ProtocolException;
+import io.orprotocol.client.Client;
+import io.orprotocol.packet.Packet;
+
+public class CallManager {
+
+    private List<CallSession> callSessions = new ArrayList<>();
+    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+    private static final long RINGING_TIMEOUT = 30 * 1000;
+    private ClientManager clientManager;
+
+
+    public CallManager(ClientManager clientManager) {
+        this.clientManager = clientManager;
+        scheduler.scheduleAtFixedRate(this::cleanupCallSessions, 0, 1, TimeUnit.SECONDS);
+    }
+
+    public void cleanupCallSessions() {
+        /**
+         * Такая конструкция нужна для избежания ConcurrentModificationException, 
+         * так как мы не можем удалять элементы из списка, по которому проходим в цикле, 
+         * поэтому мы сначала собираем сессии звонков, которые нужно удалить, а потом 
+         * удаляем их из основного списка
+         */
+        List<CallSession> sessionsToRemove = new ArrayList<>();
+        for (CallSession session : this.callSessions) {
+            if (session.shouldRemove()) {
+                /**
+                 * Отправляем всем в сессии что звонок завершился, так как он устарел, и удаляем сессию из списка активных сессий
+                 */
+                Packet26SignalPeer rtout = new Packet26SignalPeer();
+                rtout.setSignalType(NetworkSignalType.RINGING_TIMEOUT);
+                Packet26SignalPeer endCallPacket = new Packet26SignalPeer();
+                endCallPacket.setSignalType(NetworkSignalType.END_CALL);
+                endCallPacket.setJoinToken(session.getJoinToken());
+                endCallPacket.setCallId(session.getCallId());
+
+                try {
+                    session.sendPacket(rtout, null);
+                    this.sendPacketToRinging(session, endCallPacket);
+                } catch (ProtocolException e) {
+                    e.printStackTrace();
+                }
+                sessionsToRemove.add(session);
+            }
+        }
+        for (CallSession session : sessionsToRemove) {
+            this.callSessions.remove(session);
+        }
+    }
+
+    public CallSession createCall(String callId, String joinToken) {
+        CallSession session = new CallSession(callId, joinToken, RINGING_TIMEOUT);
+        this.callSessions.add(session);
+        return session;
+    }
+
+    public CallSession getCallSession(String callId, String joinToken) {
+        for (CallSession session : this.callSessions) {
+            if (session.getCallId().equals(callId) && session.getJoinToken().equals(joinToken)) {
+                return session;
+            }
+        }
+        return null;
+    }
+
+    public CallSession getCallSession(Room room) {
+        for (CallSession session : this.callSessions) {
+            if (session.getRoom() != null && session.getRoom().equals(room)) {
+                return session;
+            }
+        }
+        return null;
+    }
+
+    public boolean isBusy(String publicKey) {
+        for (CallSession session : this.callSessions) {
+            if (session.clients.containsKey(publicKey)) {
+                return true;
+            }
+            if(session.ringing.containsKey(publicKey)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public void sendPacketToRinging(CallSession session, Packet packet) throws ProtocolException {
+        for (String publicKey : session.ringing.keySet()) {
+            this.clientManager.sendPacketToAuthorizedPK(publicKey, packet);
+        }
+    }
+
+    public CallSession getCallSession(Client client) {
+        for (CallSession session : this.callSessions) {
+            if (session.clients.containsValue(client)) {
+                return session;
+            }
+        }
+        return null;
+    }
+
+    public void removeSession(CallSession session) {
+        this.callSessions.remove(session);
+    }
+
+}

src/main/java/im/rosetta/calls/CallSession.java

@@ -0,0 +1,131 @@
+package im.rosetta.calls;
+
+import java.util.HashMap;
+
+import io.g365sfu.Room;
+import io.orprotocol.ProtocolException;
+import io.orprotocol.client.Client;
+import io.orprotocol.packet.Packet;
+
+public class CallSession {
+
+    public String callId;
+    public String joinToken;
+    public Long createdAt;
+    public Long ringingTimeout;
+    /**
+     * Клиенты которым сейчас идет дозвон (публичные ключи)
+     * Клиенты в этом списке не могут принимать другие звонки, так как они уже заняты дозвоном,
+     * но они еще не в звонке, так как не приняли звонок
+     * 
+     * Клиенты удаляются из этого списка, когда они принимают звонок или отклоняют его, 
+     * тогда они либо переходят в звонок, либо становятся свободными для других звонков
+     * pk -> время начала дозвона (timestamp в миллисекундах)
+     */
+    public HashMap<String, Long> ringing;
+    /**
+     * Клиенты, которые уже приняли звонок и находятся в звонке (публичные ключи) и их сокеты
+     * pk -> client
+     */
+    public HashMap<String, Client> clients;
+    /**
+     * Если звонок активен у него появляется комната, иначе комната null
+     */
+    public Room room;
+    
+
+    public CallSession(String callId, String joinToken, Long ringingTimeout) {
+        this.callId = callId;
+        this.joinToken = joinToken;
+        this.clients = new HashMap<>();
+        this.ringing = new HashMap<>();
+        this.createdAt = System.currentTimeMillis();
+        this.ringingTimeout = ringingTimeout;
+    }
+
+    public void joinCall(String publicKey, Client client) {
+        if(this.ringing.containsKey(publicKey)) {
+            this.ringing.remove(publicKey);
+        }
+        this.clients.put(publicKey, client);
+    }
+
+    /**
+     * Проверяет, может ли этот публичный ключ выполнять какие-либо действия в рамках этой сессии звонка,
+     * чтобы не допустить выполнение действий от посторонних публичных ключей, которые не участвуют в звонке
+     * @param publicKey Публичный ключ для проверки
+     * @return true, если этот публичный ключ может выполнять действия в рамках этой сессии звонка, false иначе
+     */
+    public boolean isValidSource(String publicKey) {
+        return this.ringing.containsKey(publicKey) || this.clients.containsKey(publicKey);
+    }
+
+    /**
+     * Получаем публичный ключ клиента по его сокету, чтобы понимать, кто отправляет сигналы в рамках звонка
+     * @param client Сокет клиента, для которого нужно получить публичный ключ
+     * @return Публичный ключ клиента, или null если клиент не найден в сессии звонка
+     */
+    public String getPublicKey(Client client) {
+        for (String publicKey : this.clients.keySet()) {
+            if (this.clients.get(publicKey).equals(client)) {
+                return publicKey;
+            }
+        }
+        return null;
+    }
+
+    public String getCallId() {
+        return this.callId;
+    }
+
+    public String getJoinToken() {
+        return this.joinToken;
+    }
+
+    public void leaveCall(String publicKey) {
+        if(this.clients.containsKey(publicKey)) {
+            this.clients.remove(publicKey);
+        }
+        if(this.ringing.containsKey(publicKey)) {
+            this.ringing.remove(publicKey);
+        }
+    }
+
+    public void addRinging(String publicKey) {
+        this.ringing.put(publicKey, System.currentTimeMillis());
+    }
+
+    public void setRoom(Room room) {
+        this.room = room;
+    }
+
+    public Room getRoom() {
+        return this.room;
+    }
+
+    /**
+     * Отправляем пакет всем участникам звонка, кроме исключенного клиента (обычно отправителя)
+     * @param packet Пакет для отправки
+     * @param excludeClient Клиент, которому не нужно отправлять пакет (обычно отправитель)
+     * @throws ProtocolException Если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacket(Packet packet, Client excludeClient) throws ProtocolException {
+        for (Client client : this.clients.values()) {
+            if (!client.equals(excludeClient)) {
+                client.send(packet);
+            }
+        }
+    }
+
+    /**
+     * Проверяем, нужно ли удалять сессию звонка. 
+     * Сессию звонка нужно удалять, если в ней меньше 2 клиентов и при этом нет клиентов в состоянии дозвона, 
+     * или если сессия была создана более 1 минуты назад, так как это значит, что клиенты не ответили на звонок и он устарел
+     * @return true, если сессию звонка нужно удалять, false иначе
+     */
+    public boolean shouldRemove() {
+        return this.clients.size() <= 1 && (this.ringing.size() == 0 || System.currentTimeMillis() - this.createdAt > this.ringingTimeout);
+    }
+
+    
+}

src/main/java/im/rosetta/client/ClientManager.java

@@ -56,9 +56,10 @@ public class ClientManager {
      * Отправить пакет ВСЕМ АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
      * @param publicKey публичный ключ получателя
      * @param packet пакет для отправки
+     * @param exclude клиент, который не должен получать этот пакет, может быть null
      * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
      */
-    public void sendPacketToAuthorizedPK(String publicKey, Packet packet) throws ProtocolException {
+    public void sendPacketToAuthorizedPK(String publicKey, Packet packet, Client exclude) throws ProtocolException {
         Set<Client> clients = this.clientIndexer.getClients(ECIAuthentificate.class, "publicKey", publicKey);
         if(clients == null){
             /**
@@ -77,12 +78,55 @@ public class ClientManager {
                 continue;
             }
             /**
-             * Отправляем пакет каждому клиенту с таким публичным ключом (то есть всем его авторизованным сессиям/устройствам)
+             * Отправляем пакет каждому клиенту с таким публичным ключом (то есть всем его авторизованным сессиям/устройствам),
+             * исключая клиента exclude
              */
+            if(exclude != null && client.equals(exclude)){
+                /**
+                 * Этот клиент является исключением, он не должен получать этот пакет
+                 */
+                continue;
+            }
             client.send(packet);
         }
     }
 
+    /**
+     * Отправить пакет ВСЕМ АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
+     * @param publicKey публичный ключ получателя
+     * @param packet пакет для отправки
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(String publicKey, Packet packet) throws ProtocolException {
+        this.sendPacketToAuthorizedPK(publicKey, packet, null);
+    }
+
+    /**
+     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
+     * @param publicKeys список публичных ключей получателей
+     * @param packet пакет для отправки
+     * @param exclude клиент, который не должен получать этот пакет, может быть null
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet, Client exclude) throws ProtocolException {
+        for(String publicKey : publicKeys){
+            this.sendPacketToAuthorizedPK(publicKey, packet, exclude);
+        }
+    }
+
+    /**
+     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
+     * @param publicKeys список публичных ключей получателей
+     * @param packet пакет для отправки
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet) throws ProtocolException {
+        for(String publicKey : publicKeys){
+            this.sendPacketToAuthorizedPK(publicKey, packet, null);
+        }
+    }
+
+
     /**
      * Отправить пакет всем клиентам с публичными ключом как у client, кроме клиента client, который является отправителем и не должен получать этот пакет
      * @param client клиент
@@ -111,18 +155,6 @@ public class ClientManager {
         }
     }
 
-    /**
-     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
-     * @param publicKeys список публичных ключей получателей
-     * @param packet пакет для отправки
-     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
-     */
-    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet) throws ProtocolException {
-        for(String publicKey : publicKeys){
-            this.sendPacketToAuthorizedPK(publicKey, packet);
-        }
-    }
-
     /**
      * Получить список клиентов по публичному ключу (get PublicKey clients), могут быть неавторизованные клиенты
      * @param publicKey публичный ключ клиента

src/main/java/im/rosetta/database/entity/Device.java

@@ -1,40 +1,51 @@
 package im.rosetta.database.entity;
 
-import im.rosetta.database.CreateUpdateEntity;
+import java.util.ArrayList;
+import java.util.List;
 
+import im.rosetta.database.CreateUpdateEntity;
+import jakarta.persistence.CascadeType;
 import jakarta.persistence.Column;
 import jakarta.persistence.Entity;
+import jakarta.persistence.FetchType;
 import jakarta.persistence.GeneratedValue;
 import jakarta.persistence.GenerationType;
 import jakarta.persistence.Id;
 import jakarta.persistence.Index;
+import jakarta.persistence.OneToMany;
 import jakarta.persistence.Table;
 
 @Entity
 @Table(name = "devices", indexes = {
     @Index(name = "idx_public_key", columnList = "publicKey, deviceId", unique = true)
 })
-
 public class Device extends CreateUpdateEntity {
 
     @Id
-	@GeneratedValue(strategy = GenerationType.IDENTITY)
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
     private Long id;
 
     @Column(name = "publicKey", nullable = false)
     private String publicKey;
+
     @Column(name = "deviceId", nullable = false)
     private String deviceId;
+
     @Column(name = "deviceName", nullable = false)
     private String deviceName;
+
     @Column(name = "deviceOs", nullable = false)
     private String deviceOs;
+
     /**
      * Время завершения сессии устройства
      */
     @Column(name = "syncTime", nullable = true, columnDefinition = "bigint default 0")
     private Long syncTime;
 
+    @OneToMany(mappedBy = "device", cascade = CascadeType.ALL, orphanRemoval = true, fetch = FetchType.LAZY)
+    private List<PushToken> tokens = new ArrayList<>();
+
     public Long getId() {
         return id;
     }
@@ -59,6 +70,10 @@ public class Device extends CreateUpdateEntity {
         return syncTime;
     }
 
+    public List<PushToken> getTokens() {
+        return tokens;
+    }
+
     public void setSyncTime(Long syncTime) {
         this.syncTime = syncTime;
     }
@@ -79,4 +94,24 @@ public class Device extends CreateUpdateEntity {
         this.deviceOs = deviceOs;
     }
 
-}
+    public void setTokens(List<PushToken> tokens) {
+        this.tokens = tokens;
+    }
+
+    public void addToken(PushToken token) {
+        if (token == null) {
+            return;
+        }
+        this.tokens.add(token);
+        token.setDevice(this);
+    }
+
+    public void removeToken(PushToken token) {
+        if (token == null) {
+            return;
+        }
+        this.tokens.remove(token);
+        token.setDevice(null);
+    }
+
+}

src/main/java/im/rosetta/database/entity/PushToken.java

@@ -0,0 +1,85 @@
+package im.rosetta.database.entity;
+
+import im.rosetta.database.CreateUpdateEntity;
+import im.rosetta.packet.runtime.TokenType;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.FetchType;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Index;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+
+@Entity
+@Table(
+    name = "device_tokens",
+    uniqueConstraints = {
+        @UniqueConstraint(name = "uq_device_token", columnNames = {"device_id", "type", "token"})
+    },
+    indexes = {
+        @Index(name = "idx_device_token_type", columnList = "type"),
+        @Index(name = "idx_device_token_token", columnList = "token")
+    }
+)
+public class PushToken extends CreateUpdateEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @ManyToOne(optional = false, fetch = FetchType.LAZY)
+    @JoinColumn(name = "device_id", nullable = false)
+    private Device device;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "type", nullable = false, length = 32)
+    private TokenType type;
+
+    @Column(name = "token", nullable = false, columnDefinition = "TEXT")
+    private String token;
+
+    public Long getId() {
+        return id;
+    }
+
+    public Device getDevice() {
+        return device;
+    }
+
+    public TokenType getType() {
+        return type;
+    }
+
+    public String getToken() {
+        return token;
+    }
+
+    public void setDevice(Device device) {
+        this.device = device;
+    }
+
+    public void setType(TokenType type) {
+        this.type = type;
+    }
+
+    public void setToken(String token) {
+        this.token = token;
+    }
+
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        PushToken pushToken = (PushToken) o;
+
+        if (!device.equals(pushToken.device)) return false;
+        if (type != pushToken.type) return false;
+        return token.equals(pushToken.token);
+    }
+}

src/main/java/im/rosetta/database/entity/User.java

@@ -1,10 +1,8 @@
 package im.rosetta.database.entity;
 
 import im.rosetta.database.CreateUpdateEntity;
-import im.rosetta.database.converters.StringListConverter;
 
 import jakarta.persistence.Column;
-import jakarta.persistence.Convert;
 import jakarta.persistence.Entity;
 import jakarta.persistence.GeneratedValue;
 import jakarta.persistence.GenerationType;
@@ -12,9 +10,6 @@ import jakarta.persistence.Id;
 import jakarta.persistence.Index;
 import jakarta.persistence.Table;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @Entity
 @Table(name = "users", indexes = {
 		@Index(name = "idx_users_publickey", columnList = "publicKey", unique = true)
@@ -40,10 +35,6 @@ public class User extends CreateUpdateEntity {
 	@Column(name = "publicKey", nullable = false, unique = true)
 	private String publicKey;
 
-	@Convert(converter = StringListConverter.class)
-	@Column(name = "notificationsTokens", nullable = false, columnDefinition = "TEXT")
-	private List<String> notificationsTokens = new ArrayList<>();
-
 
 	public Long getId() {
 		return id;
@@ -89,12 +80,4 @@ public class User extends CreateUpdateEntity {
 		this.verified = verified;
 	}
 
-	public List<String> getNotificationsTokens() {
-		return notificationsTokens;
-	}
-
-	public void setNotificationsTokens(List<String> notificationsTokens) {
-		this.notificationsTokens = notificationsTokens;
-	}
-
 }

src/main/java/im/rosetta/database/repository/PushTokenRepository.java

@@ -0,0 +1,59 @@
+package im.rosetta.database.repository;
+
+import java.util.List;
+
+import org.hibernate.Session;
+
+import im.rosetta.database.HibernateUtil;
+import im.rosetta.database.Repository;
+import im.rosetta.database.entity.Device;
+import im.rosetta.database.entity.PushToken;
+
+public class PushTokenRepository extends Repository<PushToken> {
+
+    public PushTokenRepository() {
+        super(PushToken.class);
+    }
+
+    /**
+     * Найти все токены для заданного публичного ключа
+     * @param publicKey публичный ключ пользователя
+     * @return список токенов для заданного публичного ключа
+     */
+    public List<PushToken> findTokens(String publicKey) {
+        try (Session session = HibernateUtil.getSessionFactory().openSession()) {
+            return session.createQuery("SELECT pt FROM PushToken pt JOIN pt.device d WHERE d.publicKey = :publicKey", PushToken.class)
+                    .setParameter("publicKey", publicKey)
+                    .getResultList();
+        }
+    }
+
+    /**
+     * Найти все токены для заданного списка публичных ключей
+     * @param publicKeys список публичных ключей пользователей
+     * @return список токенов для заданного списка публичных ключей
+     */
+    public List<PushToken> findTokens(List<String> publicKeys) {
+        try (Session session = HibernateUtil.getSessionFactory().openSession()) {
+            return session.createQuery("SELECT pt FROM PushToken pt JOIN pt.device d WHERE d.publicKey IN :publicKeys", PushToken.class)
+                    .setParameter("publicKeys", publicKeys)
+                    .getResultList();
+        }
+    }
+
+    /**
+     * Найти токен для заданного устройства и токена
+     * @param device устройство, которому принадлежит токен
+     * @param token токен, который нужно найти
+     * @return токен для заданного устройства и токена, или null, если такой токен не найден
+     */
+    public PushToken findToken(Device device, String token){
+        try (Session session = HibernateUtil.getSessionFactory().openSession()) {
+            return session.createQuery("SELECT pt FROM PushToken pt WHERE pt.device = :device AND pt.token = :token", PushToken.class)
+                    .setParameter("device", device)
+                    .setParameter("token", token)
+                    .uniqueResult();
+        }
+    }
+
+}

src/main/java/im/rosetta/executors/Executor16PushNotification.java

@@ -1,11 +1,16 @@
 package im.rosetta.executors;
 
+import java.util.List;
+
 import im.rosetta.Failures;
 import im.rosetta.client.tags.ECIAuthentificate;
-import im.rosetta.database.entity.User;
-import im.rosetta.database.repository.UserRepository;
+import im.rosetta.database.entity.Device;
+import im.rosetta.database.entity.PushToken;
+import im.rosetta.database.repository.DeviceRepository;
+import im.rosetta.database.repository.PushTokenRepository;
 import im.rosetta.packet.Packet16PushNotification;
-import im.rosetta.service.services.UserService;
+import im.rosetta.packet.runtime.NetworkNotificationAction;
+import im.rosetta.service.services.DeviceService;
 
 import io.orprotocol.ProtocolException;
 import io.orprotocol.client.Client;
@@ -13,8 +18,9 @@ import io.orprotocol.packet.PacketExecutor;
 
 public class Executor16PushNotification extends PacketExecutor<Packet16PushNotification> {
 
-    private final UserRepository userRepository = new UserRepository();
-    private final UserService userService = new UserService(userRepository);
+    private final DeviceRepository deviceRepository = new DeviceRepository();
+    private final DeviceService deviceService = new DeviceService(deviceRepository);
+    private final PushTokenRepository pushTokenRepository = new PushTokenRepository();
 
     @Override
     public void onPacketReceived(Packet16PushNotification packet, Client client) throws Exception, ProtocolException {
@@ -33,15 +39,50 @@ public class Executor16PushNotification extends PacketExecutor<Packet16PushNotif
              */
             return;
         }
-        User user = userService.fromClient(client);
-        switch (packet.getAction()) {
-            case SUBSCRIBE:
-                userService.subscribeToPushNotifications(user, notificationToken);
-                break;
-            case UNSUBSCRIBE:
-                userService.unsubscribeFromPushNotifications(user, notificationToken);
-                break;
+
+        Device device = this.findDevice(eciAuthentificate.getPublicKey(), packet.getDeviceId());
+        if(device == null){
+            /**
+             * Устройство не найдено, значит оно не верифицировано
+             */
+            client.disconnect();
+            return;
+        }
+        PushToken pushToken = this.pushTokenRepository.findToken(device, notificationToken);
+
+        if(packet.getAction() == NetworkNotificationAction.SUBSCRIBE && pushToken == null){
+            /**
+             * Подписка на токен только если токен еще не подписан
+             */
+            PushToken token = new PushToken();
+            token.setToken(notificationToken);
+            token.setDevice(device);
+            token.setType(packet.getTokenType());
+            this.pushTokenRepository.save(token);
+        }
+        if(packet.getAction() == NetworkNotificationAction.UNSUBSCRIBE && pushToken != null){
+            /**
+             * Отписка от токена только если токен уже подписан
+             */
+            this.pushTokenRepository.delete(pushToken);
         }
     }
 
+    private Device findDevice(String publicKey, String deviceId) {
+        List<Device> devices = this.deviceService.getDevicesByPK(publicKey);
+        if(devices.size() == 0){
+            /**
+             * У пользователя нет устройств, значит текущее устройство верифицировано
+             * такого быть не может, это избыточная проверка
+             */
+            return null;
+        } 
+        for(Device device : devices){
+            if(device.getDeviceId().equals(deviceId)){
+                return device;
+            }
+        }
+        return null;
+    }
+
 }

src/main/java/im/rosetta/executors/Executor26SignalPeer.java

@@ -1,10 +1,19 @@
 package im.rosetta.executors;
 
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
 import im.rosetta.Failures;
+import im.rosetta.calls.CallManager;
+import im.rosetta.calls.CallSession;
 import im.rosetta.client.ClientManager;
 import im.rosetta.client.tags.ECIAuthentificate;
 import im.rosetta.packet.Packet26SignalPeer;
 import im.rosetta.packet.runtime.NetworkSignalType;
+import im.rosetta.service.dispatch.push.PushNotifyDispatcher;
+import im.rosetta.service.dispatch.runtime.PushType;
 import im.rosetta.service.services.ForwardUnitService;
 import io.g365sfu.Room;
 import io.orprotocol.ProtocolException;
@@ -18,62 +27,196 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
 
     private ClientManager clientManager;
     private ForwardUnitService fus;
+    private PushNotifyDispatcher pushNotifyDispatcher = new PushNotifyDispatcher();
+    /**
+     * Сигналы, которые может выполнять только авторизованный пользователь,
+     * все сигналы, которые не входят в этот перечень, будут доступны для
+     * исполнения без авторизации.
+     */
+    private Set<NetworkSignalType> authentificatedTypes = new HashSet<>(){{
+        add(NetworkSignalType.CALL);
+    }};
 
-    public Executor26SignalPeer(ClientManager clientManager, ForwardUnitService fus) {
+    /**
+     * Менеджер звонков, который реализует весь необхоимый функционал для управления звонками, 
+     * например проверку занятости пользователя, и тд
+     */
+    private CallManager callManager;
+
+    public Executor26SignalPeer(ClientManager clientManager, ForwardUnitService fus, CallManager callManager) {
         this.clientManager = clientManager;
         this.fus = fus;
+        this.callManager = callManager;
     }
 
     @Override
     public void onPacketReceived(Packet26SignalPeer packet, Client client) throws Exception, ProtocolException {
+        String src = packet.getSrc();
+        String dst = packet.getDst();
         ECIAuthentificate eciAuthentificate = client.getTag(ECIAuthentificate.class);
-        if (eciAuthentificate == null || !eciAuthentificate.hasAuthorized()) {
+        NetworkSignalType type = packet.getSignalType();
+        if ((eciAuthentificate == null || !eciAuthentificate.hasAuthorized()) 
+            && this.authentificatedTypes.contains(type)) {
             /**
-             * Если клиент не авторизован, то мы не будем обрабатывать его сигналы на анициализацию звонка
+             * Если клиент не авторизован, то мы не будем обрабатывать его сигналы на инициализацию звонка и создание комнаты
              * и просто отключим его от сервера.
              */
             client.disconnect(Failures.HANDSHAKE_NOT_COMPLETED);
             return;
         }
-        NetworkSignalType type = packet.getSignalType();
+        if(src != null && !src.equals(eciAuthentificate.getPublicKey()) && authentificatedTypes.contains(type)) {
+            /**
+             * Если src в пакете не совпадает с авторизованным PK клиента, то это может означать, что клиент пытается
+             * отправить сигнал от другого пользователя, отключаем его от сервера.
+             */
+            client.disconnect(Failures.DATA_MISSMATCH);
+            return;
+        }
         if(type == NetworkSignalType.CALL) {
             /**
              * Инициируется звонок от src к dst, проверяем, что dst не занят другим звонком, если занят, то отправляем сигнал END_CALL_BECAUSE_BUSY обратно src
              */
-            Room room = this.fus.getRoomByParticipantId(packet.getDst());
-            if(room != null) {
+            if(this.callManager.isBusy(dst) || this.callManager.isBusy(src)) {
                 /**
                  * Получатель сигнала уже находится в другой комнате, значит он занят другим звонком, отправляем сигнал END_CALL_BECAUSE_BUSY обратно src
                  */
                 Packet26SignalPeer responsePacket = new Packet26SignalPeer();
                 responsePacket.setSignalType(NetworkSignalType.END_CALL_BECAUSE_BUSY);
-                this.clientManager.sendPacketToAuthorizedPK(packet.getSrc(), responsePacket);
+                client.send(responsePacket);
                 return;
             }
-        }
-        if(type == NetworkSignalType.CREATE_ROOM){
             /**
-             * Создается комната для звонка
+             * Генерируем CallID и JoinToken
              */
-            Room room = this.fus.createRoom();
-            room.addParticipant(packet.getSrc());
-            room.addParticipant(packet.getDst());
-            packet.setRoomId(room.getRoomId());
+            String callId = UUID.randomUUID().toString();
+            String joinToken = UUID.randomUUID().toString();
+            packet.setJoinToken(joinToken);
+            packet.setCallId(callId);
             /**
-             * Результат создания комнаты транслируем обоим участникам, чтобы они могли начать обмен WebRTC SDP, и тд
+             * Создаем сессию звонка и добавляем в нее звонящего
              */
-            this.clientManager.sendPacketToAuthorizedPK(packet.getSrc(), packet);
-            this.clientManager.sendPacketToAuthorizedPK(packet.getDst(), packet);
+            CallSession session = this.callManager.createCall(callId, joinToken);
+            session.joinCall(src, client);
+            /**
+             * Добавляем dst в ringing, чтобы пометить, что ему поступает звонок
+             */
+            session.addRinging(dst);
+            /**
+             * Получатель сигнала не занят, отправляем ему пуш уведомление о входящем звонке и сигнал CALL для инициализации звонка
+             */
+            pushNotifyDispatcher.sendPush(dst, new HashMap<>(){{
+                put("type", PushType.CALL);
+                put("dialog", src);
+                put("callId", callId);
+                put("joinToken", joinToken);
+            }});
+            /**
+             * Отправляем сигнал CALL всем авторизованным устройствам вызываемого абонента
+             */
+            this.clientManager.sendPacketToAuthorizedPK(dst, packet);
             return;
         }
-        /**
-         * TODO: Проверка на существование получателя
-         */
-        this.clientManager.sendPacketToAuthorizedPK(packet.getDst(), packet);
-        /**
-         * TODO: Высокоприоритетный пуш для сигналов звонков, чтобы мобильные устройства могли показать
-         * интерфейс входящего звонка, даже если приложение находится в фоне
-         */
+        if(type == NetworkSignalType.ACCEPT){
+            String callId = packet.getCallId();
+            String joinToken = packet.getJoinToken();
+            CallSession session = this.callManager.getCallSession(callId, joinToken);
+            if(session == null) {
+                /**
+                 * Сессии звонка нет
+                 */
+                client.disconnect(Failures.NO_CALL_SESSION);
+                return;
+            }
+            if(!session.isValidSource(src)) {
+                /**
+                 * Клиент не состоит в сессии звонка, отключаем его от сервера, так как он отправляет некорректные данные
+                 */
+                client.disconnect(Failures.DATA_MISSMATCH);
+                return;
+            }
+            Room room = this.fus.createRoom();
+            session.setRoom(room);
+            session.joinCall(src, client);
+            for(String participant : session.clients.keySet()) {
+                room.addParticipant(participant);
+            }
+            session.sendPacket(packet, client);
+            /**
+             * Сбрасываем вызов на всех остальных устройствах пользователя, который принимает звонок,
+             * чтобы он не смог принять или отклонить звонок с другого устройства
+             */
+            Packet26SignalPeer endCallOtherDevices = new Packet26SignalPeer();
+            endCallOtherDevices.setSignalType(NetworkSignalType.END_CALL);
+            this.clientManager.sendPacketToAuthorizedPK(src, endCallOtherDevices, client);
+            return;
+        }
+        if(type == NetworkSignalType.KEY_EXCHANGE){
+            /**
+             * Ретранслируем ключи в рамках сессии
+             */
+            CallSession session = this.callManager.getCallSession(client);
+            if(session == null) {
+                /**
+                 * Сессии звонка нет
+                 */
+                client.disconnect(Failures.NO_CALL_SESSION);
+                return;
+            }
+            /**
+             * Обмениваемся ключами в рамках сессии, ретранслируя их всем участникам сессии, кроме отправителя
+             */
+            session.sendPacket(packet, client);
+            return;
+        }
+        if(type == NetworkSignalType.END_CALL) {
+            /**
+             * Ретранслируем сигнал окончания звонка всем участникам сессии, кроме отправителя, и удаляем сессию
+             */
+            /**
+             * Сначала получаем сессию по сокету отправителя пакета, если не находим, то пробуем найти сессию по callId и joinToken из пакета, если не находим, 
+             * то отключаем клиента от сервера, так как он отправляет некорректные данные
+             */
+            CallSession session = this.callManager.getCallSession(client);
+            if(session == null) {
+                String callId = packet.getCallId();
+                String joinToken = packet.getJoinToken();
+                session = this.callManager.getCallSession(callId, joinToken);
+            }
+            if(session == null) {
+                /**
+                 * Сессии звонка нет, скорее всего она была удалена при обрыве RTC Peer Connection,
+                 * при срабатывании RTCPeerConnection::close на клиенте раньше, чем клиент отправил сигнал END_CALL
+                 */
+                return;
+            }
+            /**
+             * Отправляем сигнал окончания звонка всем участникам сессии, кроме отправителя
+             */
+            session.sendPacket(packet, client);
+            /**
+             * Отправляем пакет вызываемым (ringing) пользователям (которые еще не в сессии)
+             */
+            this.callManager.sendPacketToRinging(session, packet);
+            /**
+             * Удаляем сессию из активных сессий звонков
+             */
+            this.callManager.removeSession(session);
+            return;
+        }
+        if(type == NetworkSignalType.ACTIVE) {
+            /**
+             * Клиент сообщил, что прошел стадию обмена ключами и звонок активен
+             */
+            CallSession session = this.callManager.getCallSession(client);
+            if(session == null) {
+                /**
+                 * Сессии звонка нет
+                 */
+                client.disconnect(Failures.NO_CALL_SESSION);
+                return;
+            }
+            session.sendPacket(packet, null);
+        }
     }
 
 }

src/main/java/im/rosetta/executors/Executor27WebRTC.java

@@ -1,10 +1,10 @@
 package im.rosetta.executors;
 
 import im.rosetta.Failures;
-import im.rosetta.client.tags.ECIAuthentificate;
+import im.rosetta.calls.CallManager;
+import im.rosetta.calls.CallSession;
 import im.rosetta.packet.Packet27WebRTC;
 import im.rosetta.packet.runtime.NetworkWebRTCType;
-import im.rosetta.service.services.ForwardUnitService;
 import io.g365sfu.Room;
 import io.orprotocol.ProtocolException;
 import io.orprotocol.client.Client;
@@ -12,35 +12,42 @@ import io.orprotocol.packet.PacketExecutor;
 
 public class Executor27WebRTC extends PacketExecutor<Packet27WebRTC> {
 
-    private ForwardUnitService fus;
+    private CallManager callManager;
 
-    public Executor27WebRTC(ForwardUnitService fus) {
-        this.fus = fus;
+    public Executor27WebRTC(CallManager callManager) {
+        this.callManager = callManager;
     }
 
     @Override
     public void onPacketReceived(Packet27WebRTC packet, Client client) throws Exception, ProtocolException {
-        ECIAuthentificate eciAuthentificate = client.getTag(ECIAuthentificate.class);
-        if(eciAuthentificate == null || !eciAuthentificate.hasAuthorized()) {
+        /**
+         * Получаем, в какой сессии находится этот сокет
+         */
+        CallSession session = this.callManager.getCallSession(client);
+        if(session == null) {
             /**
-             * Если клиент не авторизован, то мы не будем обрабатывать его сигналы на инициализацию звонка
-             * и просто отключим его от сервера.
+             * Если сессия не найдена, то мы не будем обрабатывать сигналы для звонка
              */
-            client.disconnect(Failures.HANDSHAKE_NOT_COMPLETED);
             return;
         }
-
-        String publicKey = eciAuthentificate.getPublicKey();
-
-        /**
-         * Так как в комнатах Participants это публичные ключи пользователей, то мы можем
-         * найти комнату, в которой находится пользователь, по его публичному ключу
-         */
-        Room room = this.fus.getRoomByParticipantId(publicKey);
+        Room room = session.getRoom();
         if(room == null) {
             /**
-             * Если комната не найдена, то мы не будем обрабатывать сигналы для звонка
-             * и просто отключим клиента от сервера.
+             * Звонок еще не активен, а значит комнаты еще нет. Нельзя обменяться WebRTC сигналами пока комнаты еще нет.
+             */
+            return;
+        }
+        /**
+         * Получаем публичный ключ, которым представился клиент, в рамках сессии звонка.
+         * Мы не делаем это через ECIAuthentificate, так как в рамках звонка клиент может не быть авторизован, но при этом он уже находится в сессии звонка, и мы можем идентифицировать его по публичному ключу, 
+         * который он указал при присоединении к звонку используя joinToken.
+         * Так что, несмотря на то, что клиент может быть не авторизован, мы все равно можем достоверно знать его публичный ключ
+         */
+        String publicKey = session.getPublicKey(client);
+        if(publicKey == null) {
+            /**
+             * Избыточная проверка, так как если клиент находится в сессии, то он должен быть в списке клиентов сессии, 
+             * но на всякий случай проверим это, чтобы избежать возможных ошибок
              */
             client.disconnect(Failures.DATA_MISSMATCH);
             return;

src/main/java/im/rosetta/executors/Executor28IceServers.java

@@ -2,8 +2,6 @@ package im.rosetta.executors;
 
 import java.util.ArrayList;
 
-import im.rosetta.Failures;
-import im.rosetta.client.tags.ECIAuthentificate;
 import im.rosetta.packet.Packet28IceServers;
 import im.rosetta.service.services.ForwardUnitService;
 import io.orprotocol.ProtocolException;
@@ -20,15 +18,6 @@ public class Executor28IceServers extends PacketExecutor<Packet28IceServers> {
 
     @Override
     public void onPacketReceived(Packet28IceServers packet, Client client) throws Exception, ProtocolException {
-        ECIAuthentificate eciAuthentificate = client.getTag(ECIAuthentificate.class);
-        if(eciAuthentificate == null || !eciAuthentificate.hasAuthorized()) {
-            /**
-             * Если клиент не авторизован, то мы не будем обрабатывать его запрос на получение ICE серверов
-             * и просто отключим его от сервера.
-             */
-            client.disconnect(Failures.HANDSHAKE_NOT_COMPLETED);
-            return;
-        }
         /**
          * Берем TURN сервера и отправляем их клиенту
          */

src/main/java/im/rosetta/packet/Packet16PushNotification.java

@@ -1,7 +1,7 @@
 package im.rosetta.packet;
 
 import im.rosetta.packet.runtime.NetworkNotificationAction;
-
+import im.rosetta.packet.runtime.TokenType;
 import io.orprotocol.Stream;
 import io.orprotocol.packet.Packet;
 
@@ -12,11 +12,15 @@ public class Packet16PushNotification extends Packet {
 
     private String notificationToken;
     private NetworkNotificationAction action;
+    private TokenType tokenType;
+    private String deviceId;
 
     @Override
     public void read(Stream stream) {
         this.notificationToken = stream.readString();
         this.action = NetworkNotificationAction.fromCode(stream.readInt8());
+        this.tokenType = TokenType.fromCode(stream.readInt8());
+        this.deviceId = stream.readString();
     }
 
     @Override
@@ -25,6 +29,8 @@ public class Packet16PushNotification extends Packet {
         stream.writeInt16(this.packetId);
         stream.writeString(notificationToken);
         stream.writeInt8(action.getCode());
+        stream.writeInt8(tokenType.getCode());
+        stream.writeString(deviceId);
         return stream;
     }
 
@@ -60,4 +66,36 @@ public class Packet16PushNotification extends Packet {
         this.action = action;
     }
 
-}
+    /**
+     * Устанавливает тип токена пуш уведомлений.
+     * @param tokenType тип токена пуш уведомлений
+     */
+    public void setTokenType(TokenType tokenType) {
+        this.tokenType = tokenType;
+    }
+
+    /**
+     * Получить тип токена пуш уведомлений, который нужно подписать или отписать в зависимости от action
+     * @return тип токена пуш уведомлений
+     */
+    public TokenType getTokenType() {
+        return tokenType;
+    }
+
+    /**
+     * Девайс которому принадлежит токен пуш уведомлений, который нужно подписать или отписать в зависимости от action
+     * @return
+     */
+    public String getDeviceId() {
+        return deviceId;
+    }
+
+    /**
+     * Устанавливает девайс которому принадлежит токен пуш уведомлений, который нужно подписать или отписать в зависимости от action
+     * @param deviceId девайс которому принадлежит токен пуш уведомлений
+     */
+    public void setDeviceId(String deviceId) {
+        this.deviceId = deviceId;
+    }
+
+}

src/main/java/im/rosetta/packet/Packet26SignalPeer.java

@@ -29,16 +29,18 @@ public class Packet26SignalPeer extends Packet {
      * Тип сигнала
      */
     private NetworkSignalType signalType;
-    
     /**
-     * Идентификатор комнаты, в которой происходит звонок, заполняется если тип сигнала CREATE_ROOM, иначе null
+     * callId и joinToken нужны для того, чтобы идентифицировать сессию звонка. Так как roomId это только ID комнаты на sfu
      */
-    private String roomId;
+    private String callId;
+    private String joinToken;
 
     @Override
     public void read(Stream stream) {
         this.signalType = NetworkSignalType.fromCode(stream.readInt8());
-        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED) {
+        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY 
+            || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED
+            || this.signalType == NetworkSignalType.RINGING_TIMEOUT) {
             return;
         }
         this.src = stream.readString();
@@ -46,8 +48,9 @@ public class Packet26SignalPeer extends Packet {
         if (signalType == NetworkSignalType.KEY_EXCHANGE) {
             this.sharedPublic = stream.readString();
         }
-        if(signalType == NetworkSignalType.CREATE_ROOM) {
-            this.roomId = stream.readString();
+        if(signalType == NetworkSignalType.CALL || signalType == NetworkSignalType.ACCEPT || signalType == NetworkSignalType.END_CALL) {
+            this.callId = stream.readString();
+            this.joinToken = stream.readString();
         }
     }
 
@@ -56,7 +59,9 @@ public class Packet26SignalPeer extends Packet {
         Stream stream = new Stream();
         stream.writeInt16(this.packetId);
         stream.writeInt8(this.signalType.getCode());
-        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED) {
+        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY 
+            || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED
+            || this.signalType == NetworkSignalType.RINGING_TIMEOUT) {
             return stream;
         }
         stream.writeString(this.src);
@@ -64,8 +69,9 @@ public class Packet26SignalPeer extends Packet {
         if (signalType == NetworkSignalType.KEY_EXCHANGE) {
             stream.writeString(this.sharedPublic);
         }
-        if(signalType == NetworkSignalType.CREATE_ROOM) {
-            stream.writeString(this.roomId);
+        if(signalType == NetworkSignalType.CALL || signalType == NetworkSignalType.ACCEPT || signalType == NetworkSignalType.END_CALL) {
+            stream.writeString(this.callId);
+            stream.writeString(this.joinToken);
         }
         return stream;
     }
@@ -134,20 +140,35 @@ public class Packet26SignalPeer extends Packet {
     public void setSignalType(NetworkSignalType signalType) {
         this.signalType = signalType;
     }
+    /**
+     * Получить идентификатор сессии звонка, если тип сигнала CALL или ACCEPT
+     * @return идентификатор сессии звонка, если тип сигнала CALL или ACCEPT, иначе null
+     */
+    public String getCallId() {
+        return callId;
+    }
 
     /**
-     * Получить идентификатор созданной комнаты, если тип сигнала CREATE_ROOM
-     * @return идентификатор комнаты, если тип сигнала CREATE_ROOM, иначе null
+     * Установить идентификатор сессии звонка, если тип сигнала CALL или ACCEPT
+     * @param callId идентификатор сессии звонка, если тип сигнала CALL или ACCEPT
      */
-    public String getRoomId() {
-        return roomId;
+    public void setCallId(String callId) {
+        this.callId = callId;
     }
 
-     /**
-     * Установить идентификатор комнаты, в которой происходит звонок, если тип сигнала CREATE_ROOM
-     * @param roomId идентификатор комнаты, если тип сигнала CREATE_ROOM
+    /**
+     * Получить токен для присоединения к сессии звонка, если тип сигнала CALL или ACCEPT
+     * @return токен для присоединения к сессии звонка, если тип сигнала CALL или ACCEPT, иначе null
      */
-    public void setRoomId(String roomId) {
-        this.roomId = roomId;
+    public String getJoinToken() {
+        return joinToken;
+    }
+
+    /**
+     * Установить токен для присоединения к сессии звонка, если тип сигнала CALL или ACCEPT
+     * @param joinToken токен для присоединения к сессии звонка, если тип сигнала CALL или ACCEPT
+     */
+    public void setJoinToken(String joinToken) {
+        this.joinToken = joinToken;
     }
 }

src/main/java/im/rosetta/packet/Packet27WebRTC.java

@@ -60,6 +60,4 @@ public class Packet27WebRTC extends Packet {
     public void setType(NetworkWebRTCType type) {
         this.type = type;
     }
-
-
 }

src/main/java/im/rosetta/packet/runtime/AttachmentType.java

@@ -8,7 +8,9 @@ public enum AttachmentType {
     MESSAGES(1),
     FILE(2),
     AVATAR(3),
-    CALL(4);
+    CALL(4),
+    VOICE(5),
+    VIDEO_CIRCLE(6);
 
 
     private final int code;

src/main/java/im/rosetta/packet/runtime/NetworkSignalType.java

@@ -21,9 +21,9 @@ public enum NetworkSignalType {
      */
     END_CALL(3),
     /**
-     * Создание комнаты
+     * Активная стадия звонка, значит комната уже создана на SFU
      */
-    CREATE_ROOM(4),
+    ACTIVE(4),
     /**
      * Обрыв связи с пиром
      */
@@ -31,7 +31,16 @@ public enum NetworkSignalType {
     /**
      * Не удалось дозвониться - пользователь занят другим звонком
      */
-    END_CALL_BECAUSE_BUSY(6);
+    END_CALL_BECAUSE_BUSY(6),
+    /**
+     * Принятие звонка
+     */
+    ACCEPT(7),
+    /**
+     * Таймаут на этапе дозвона, если пользователь не ответил на звонок в течение определенного времени, то звонок считается неуспешным и вызывающей
+     * стороне отправляется этот сигнал, а сессия звонка удаляется, так как она уже не актуальна
+     */
+    RINGING_TIMEOUT(8);
 
     private final int code;
 

src/main/java/im/rosetta/packet/runtime/README.md

@@ -0,0 +1,3 @@
+## Именование
+
+Если в названии класса есть слово Network значит этот тип используется только при сетевых передачах, если слова Network в названии класса нет значит он используется где-то кроме сетевых передач.

src/main/java/im/rosetta/packet/runtime/TokenType.java

@@ -0,0 +1,34 @@
+package im.rosetta.packet.runtime;
+
+/**
+ * Тип токена для пуш уведомлений. Используется в Packet16PushNotification для указания типа токена, который нужно подписать/отписать.
+ */
+public enum TokenType {
+    /**
+     * FCM токен для пуш уведомлений, используется и iOS и Android
+     */
+    FCM(0),
+    /**
+     * VoIP токен для пуш уведомлений, используется только на iOS для VoIP уведомлений
+     */
+    VoIPApns(1);
+
+    private int code;
+
+    TokenType(int code){
+        this.code = code;
+    }
+
+    public int getCode() {
+        return code;
+    }
+
+    public static TokenType fromCode(int code) {
+        for (TokenType type : values()) {
+            if (type.code == code) {
+                return type;
+            }
+        }
+        throw new IllegalArgumentException("Unknown NetworkTokenType code: " + code);
+    }
+}

src/main/java/im/rosetta/service/dispatch/FirebaseDispatcher.java

@@ -1,179 +0,0 @@
-package im.rosetta.service.dispatch;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-import com.google.auth.oauth2.GoogleCredentials;
-import com.google.firebase.FirebaseApp;
-import com.google.firebase.FirebaseOptions;
-import com.google.firebase.messaging.ApnsConfig;
-import com.google.firebase.messaging.Aps;
-import com.google.firebase.messaging.ApsAlert;
-import com.google.firebase.messaging.FirebaseMessaging;
-import com.google.firebase.messaging.Message;
-import com.google.firebase.messaging.Notification;
-
-import im.rosetta.database.repository.UserRepository;
-import im.rosetta.service.services.UserService;
-
-/**
- * Класс для отправки push-уведомлений пользователям через Firebase Cloud Messaging (FCM).
- */
-public class FirebaseDispatcher {
-
-    private UserRepository userRepository = new UserRepository();
-    private UserService userService = new UserService(userRepository);
-    private final ExecutorService executor = Executors.newFixedThreadPool(10);
-
-    public FirebaseDispatcher() {
-        initializeFirebase();
-    }
-
-    /**
-     * Инициализация Firebase Admin SDK
-     */
-    private void initializeFirebase() {
-        if (FirebaseApp.getApps().isEmpty()) {
-            try {
-                String firebaseCredentialsPath = System.getenv("FIREBASE_CREDENTIALS_PATH");
-                if (firebaseCredentialsPath == null || firebaseCredentialsPath.isEmpty()) {
-                    throw new IllegalStateException("FIREBASE_CREDENTIALS_PATH environment variable is not set");
-                }
-                
-                FileInputStream serviceAccount = new FileInputStream(firebaseCredentialsPath);
-                FirebaseOptions options = FirebaseOptions.builder()
-                    .setCredentials(GoogleCredentials.fromStream(serviceAccount))
-                    .build();
-                FirebaseApp.initializeApp(options);
-            } catch (IOException e) {
-                throw new RuntimeException("Failed to initialize Firebase", e);
-            }
-        }
-    }
-
-    public void sendPushNotification(String publicKey, String title, String messageText, String senderPublicKey) {
-        executor.submit(() -> {
-            try {
-                List<String> tokens = userService.getNotificationsTokens(publicKey);
-                if (tokens == null || tokens.isEmpty()) {
-                    return;
-                }
-
-                for (String token : tokens) {
-                    try {
-                        Message message = Message.builder()
-                            .setNotification(Notification.builder()
-                                .setTitle(title)
-                                .setBody(messageText)
-                                .build())
-                            .setApnsConfig(ApnsConfig.builder()
-                                .setAps(Aps.builder()
-                                    .setMutableContent(true)  
-                                    .setSound("default")         
-                                    .setAlert(ApsAlert.builder()
-                                        .setTitle(title)
-                                        .setBody(messageText)
-                                        .build())
-                                    .build())
-                                .build())
-                            .putData("sender_public_key", senderPublicKey)
-                            .setToken(token)
-                            .build();
-
-                        FirebaseMessaging.getInstance().send(message);
-                    } catch (Exception e) {
-                        e.printStackTrace();
-                    }
-                }
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        });
-    }
-
-    /**
-     * Отправляет push-уведомление пользователю с данным публичным ключом (асинхронно)
-     * @param publicKey публичный ключ пользователя
-     * @param title заголовок уведомления
-     * @param messageText текст уведомления
-     */
-    public void sendPushNotification(String publicKey, String title, String messageText) {
-        sendPushNotification(publicKey, title, messageText, null);
-    }
-
-    /**
-     * Отправляет push-уведомление нескольким пользователям (асинхронно)
-     * @param publicKeys список публичных ключей пользователей
-     * @param title заголовок уведомления
-     * @param messageText текст уведомления
-     */
-    public void sendPushNotification(List<String> publicKeys, String title, String messageText, String senderPublicKey) {
-        executor.submit(() -> {
-            for (String publicKey : publicKeys) {
-                sendPushNotificationSync(publicKey, title, messageText, senderPublicKey);
-            }
-        });
-    }
-
-    /**
-     * Отправляет push-уведомление нескольким пользователям (асинхронно)
-     * @param publicKeys список публичных ключей пользователей
-     * @param title заголовок уведомления
-     * @param messageText текст уведомления
-     */
-    public void sendPushNotification(List<String> publicKeys, String title, String messageText) {
-        executor.submit(() -> {
-            for (String publicKey : publicKeys) {
-                sendPushNotificationSync(publicKey, title, messageText, null);
-            }
-        });
-    }
-
-    private void sendPushNotificationSync(String publicKey, String title, String messageText, String senderPublicKey) {
-        try {
-            List<String> tokens = userService.getNotificationsTokens(publicKey);
-            if (tokens == null || tokens.isEmpty()) {
-                return;
-            }
-
-            for (String token : tokens) {
-                try {
-                    Message message = Message.builder()
-                        .setNotification(Notification.builder()
-                            .setTitle(title)
-                            .setBody(messageText)
-                            .build())
-                        .setApnsConfig(ApnsConfig.builder()
-                            .setAps(Aps.builder()
-                                .setMutableContent(true)  
-                                .setSound("default")         
-                                .setAlert(ApsAlert.builder()
-                                    .setTitle(title)
-                                    .setBody(messageText)
-                                    .build())
-                                .build())
-                            .build())
-                        .putData("sender_public_key", senderPublicKey != null ? senderPublicKey : "")
-                        .setToken(token)
-                        .build();
-
-                    FirebaseMessaging.getInstance().send(message);
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
-            }
-        } catch (Exception e) {
-            // Логирование ошибки
-        }
-    }
-
-    /**
-     * Завершить работу executor при остановке приложения
-     */
-    public void shutdown() {
-        executor.shutdown();
-    }
-}

src/main/java/im/rosetta/service/dispatch/MessageDispatcher.java

@@ -1,5 +1,6 @@
 package im.rosetta.service.dispatch;
 
+import java.util.HashMap;
 import java.util.List;
 
 import im.rosetta.client.ClientManager;
@@ -11,6 +12,8 @@ import im.rosetta.database.repository.UserRepository;
 import im.rosetta.packet.Packet11Typeing;
 import im.rosetta.packet.Packet7Read;
 import im.rosetta.packet.base.PacketBaseDialog;
+import im.rosetta.service.dispatch.push.PushNotifyDispatcher;
+import im.rosetta.service.dispatch.runtime.PushType;
 import im.rosetta.service.services.BufferService;
 import im.rosetta.service.services.UserService;
 import io.orprotocol.ProtocolException;
@@ -30,7 +33,7 @@ public class MessageDispatcher {
     private final ClientManager clientManager;
     private final BufferRepository bufferRepository = new BufferRepository();
     private final BufferService bufferService;
-    private final FirebaseDispatcher firebaseDispatcher = new FirebaseDispatcher();
+    private final PushNotifyDispatcher pushNotifyDispatcher = new PushNotifyDispatcher();
     private final UserRepository userRepository = new UserRepository();
     private final UserService userService = new UserService(userRepository);
 
@@ -98,14 +101,26 @@ public class MessageDispatcher {
         }
         if(packet instanceof Packet7Read){
             /**
-             * Если это пакет прочтения, то не отправляем пуш уведомление, так как это может привести к спаму пушами при чтении сообщений
+             * Если это пакет прочтения, то отправляем тихий пуш, что диалог прочитан, отправляем тому, кто читает диалог, чтобы 
+             * клиент мог очистить пуши для этого диалога
              */
+            this.pushNotifyDispatcher.sendPush(fromPublicKey, new HashMap<>(){
+                {
+                    put("type", PushType.READ);
+                    put("dialog", toPublicKey);
+                }
+            });
             return;
         }
         /**
          * Отправляем PUSH уведомление
          */
-        this.firebaseDispatcher.sendPushNotification(groupMembersPublicKeys, "Rosetta", "New message in group", toPublicKey.replace("#group:", ""));
+        this.pushNotifyDispatcher.sendPush(groupMembersPublicKeys, new HashMap<>(){
+            {
+                put("type", PushType.GROUP_MESSAGE);
+                put("dialog", toPublicKey.replace("#group:", ""));
+            }
+        });
     }
 
     /**
@@ -156,15 +171,28 @@ public class MessageDispatcher {
         }
         if(packet instanceof Packet7Read){
             /**
-             * Если это пакет прочтения, то не отправляем пуш уведомление, 
-             * так как это может привести к спаму пушами при чтении сообщений
+             * Если это пакет прочтения, то отправляем тихий пуш, что диалог прочитан, отправляем тому, кто читает диалог, чтобы 
+             * клиент мог очистить пуши для этого диалога
              */
+            this.pushNotifyDispatcher.sendPush(fromPublicKey, new HashMap<>(){
+                {
+                    put("type", PushType.READ);
+                    put("dialog", toPublicKey);
+                    put("title", user.getTitle());
+                }
+            });
             return;
         }
         /**
          * Отправляем PUSH уведомление получателю
          */
-        this.firebaseDispatcher.sendPushNotification(toPublicKey, user.getTitle(), "New message", fromPublicKey);
+        this.pushNotifyDispatcher.sendPush(toPublicKey, new HashMap<>(){
+            {
+                put("type", PushType.PERSONAL_MESSAGE);
+                put("dialog", fromPublicKey);
+                put("title", user.getTitle());
+            }
+        });
     }    
 
     /**

src/main/java/im/rosetta/service/dispatch/push/PushNotifyDispatcher.java

@@ -0,0 +1,65 @@
+package im.rosetta.service.dispatch.push;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import im.rosetta.database.entity.PushToken;
+import im.rosetta.database.repository.PushTokenRepository;
+import im.rosetta.packet.runtime.TokenType;
+import im.rosetta.service.dispatch.push.dispatchers.FCM;
+import im.rosetta.service.dispatch.push.dispatchers.VoIPApns;
+
+public class PushNotifyDispatcher {
+
+    private final ExecutorService executor = Executors.newFixedThreadPool(10);
+    private final PushTokenRepository pushTokenRepository = new PushTokenRepository();
+
+    private final HashMap<TokenType, Pusher> pushers = new HashMap<>() {{
+        put(TokenType.FCM, new FCM());
+        put(TokenType.VoIPApns, new VoIPApns());
+    }};
+
+
+    private Pusher findPusher(TokenType tokenType){
+        return this.pushers.get(tokenType);
+    }
+    
+    /**
+     * Отправить уведомление пользователю с publicKey, используя все его токены для отправки уведомления, если таковые имеются
+     * @param publicKey публичный ключ пользователя, которому нужно отправить уведомление
+     * @param data данные уведомления
+     */
+    public void sendPush(String publicKey, HashMap<String, String> data) {
+        List<PushToken> pushTokens = this.pushTokenRepository.findTokens(publicKey);
+        executor.execute(() -> {
+            for(PushToken pushToken : pushTokens){
+                Pusher pusher = this.findPusher(pushToken.getType());
+                if(pusher != null){
+                    pusher.sendPush(pushToken.getToken(), data);
+                }
+            }
+        });
+    }
+
+    /**
+     * Отправить уведомление пользователям с publicKeys, используя все их токены для отправки уведомления, если таковые имеются
+     * @param publicKeys список публичных ключей пользователей, которым нужно отправить уведомление
+     * @param data данные уведомления
+     */
+    public void sendPush(List<String> publicKeys, HashMap<String, String> data) {
+        List<PushToken> pushTokens = this.pushTokenRepository.findTokens(publicKeys);
+        executor.execute(() -> {
+            for(PushToken pushToken : pushTokens){
+                Pusher pusher = this.findPusher(pushToken.getType());
+                    if(pusher != null){
+                        pusher.sendPush(pushToken.getToken(), data);
+                    }
+                }
+            }
+        );
+    }
+
+
+}

src/main/java/im/rosetta/service/dispatch/push/Pusher.java

@@ -0,0 +1,14 @@
+package im.rosetta.service.dispatch.push;
+
+import java.util.HashMap;
+
+public abstract class Pusher {
+
+    /**
+     * Вызывается при отправке PUSH уведомления, когда уже определен токен
+     * @param token токен для уведомления (FCM/VoIP/etc..)
+     * @param data данные уведомления
+     */
+    public abstract void sendPush(String token, HashMap<String, String> data);
+
+}

src/main/java/im/rosetta/service/dispatch/push/dispatchers/FCM.java

@@ -0,0 +1,115 @@
+package im.rosetta.service.dispatch.push.dispatchers;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.HashMap;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.firebase.FirebaseApp;
+import com.google.firebase.FirebaseOptions;
+import com.google.firebase.messaging.AndroidConfig;
+import com.google.firebase.messaging.ApnsConfig;
+import com.google.firebase.messaging.Aps;
+import com.google.firebase.messaging.FirebaseMessaging;
+import com.google.firebase.messaging.Message;
+import com.google.firebase.messaging.Notification;
+
+import im.rosetta.service.dispatch.push.Pusher;
+import im.rosetta.service.dispatch.runtime.PushType;
+
+public class FCM extends Pusher {
+
+    private boolean isInitialized = false;
+
+    public FCM() {
+        initializeFirebase();
+    }
+
+    /**
+     * Инициализация Firebase Admin SDK
+     */
+    private void initializeFirebase() {
+        if (FirebaseApp.getApps().isEmpty()) {
+            try {
+                String firebaseCredentialsPath = System.getenv("FIREBASE_CREDENTIALS_PATH");
+                if (firebaseCredentialsPath == null || firebaseCredentialsPath.isEmpty()) {
+                    throw new IllegalStateException("FIREBASE_CREDENTIALS_PATH environment variable is not set");
+                }
+                
+                FileInputStream serviceAccount = new FileInputStream(firebaseCredentialsPath);
+                FirebaseOptions options = FirebaseOptions.builder()
+                    .setCredentials(GoogleCredentials.fromStream(serviceAccount))
+                    .build();
+                FirebaseApp.initializeApp(options);
+                this.isInitialized = true;
+            } catch (IOException e) {
+                this.isInitialized = false;
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private Message buildMessage(String token, HashMap<String, String> data) {
+        String type = data.get("type");
+        if(type == null){
+            throw new IllegalArgumentException("Push notification type is required in data");
+        }
+        ApnsConfig.Builder apnsConfig = ApnsConfig.builder();
+        AndroidConfig.Builder androidConfig = AndroidConfig.builder();
+        Message.Builder messageBuilder = Message.builder()
+            .setToken(token)
+            .putAllData(data);
+        switch(type) {
+            case PushType.READ:
+                /**
+                 * Тихий тип уведомления для очистки отправленных уведомлений на устройстве, 
+                 * не должен отображаться пользователю, поэтому не задаем звук и ставим contentAvailable для iOS и high priority для Android
+                 */
+                apnsConfig.setAps(Aps.builder().setContentAvailable(true).setMutableContent(true).build());
+                androidConfig.setPriority(AndroidConfig.Priority.HIGH);
+                messageBuilder.setApnsConfig(apnsConfig.build());
+                messageBuilder.setNotification(Notification.builder().setTitle("").setBody("").build());
+                messageBuilder.setAndroidConfig(androidConfig.build());
+                break;
+            case PushType.PERSONAL_MESSAGE:
+            case PushType.GROUP_MESSAGE:
+                /**
+                 * Уведомление о новом сообщении, должно отображаться пользователю, поэтому задаем звук и high priority для Android
+                 */
+                String body = type == PushType.PERSONAL_MESSAGE ? "New message" : "New group message";
+                apnsConfig.setAps(Aps.builder().setSound("default").setMutableContent(true).build());
+                androidConfig.setPriority(AndroidConfig.Priority.HIGH);
+                messageBuilder.setApnsConfig(apnsConfig.build());
+                messageBuilder.setNotification(Notification.builder().setTitle(
+                    data.getOrDefault("title", "Rosetta")
+                ).setBody(body).build());
+                messageBuilder.setAndroidConfig(androidConfig.build());
+                break;
+            case PushType.CALL:
+                /**
+                 * Это только для Android, для iOS используется VoIP APNs с отдельным сертификатом
+                 */
+                androidConfig.setPriority(AndroidConfig.Priority.HIGH);
+                messageBuilder.setAndroidConfig(androidConfig.build());
+                break;
+        }
+        return messageBuilder.build();
+    }
+
+    @Override
+    public void sendPush(String token, HashMap<String, String> data) {
+        if(!this.isInitialized){
+            /**
+             * Firebase не инициализирован, пропускаем отправку уведомлений
+             */
+            return;
+        }
+        try{
+            Message message = this.buildMessage(token, data);
+            FirebaseMessaging.getInstance().send(message);
+        }catch(Exception e){
+            
+        }
+    }
+
+}

src/main/java/im/rosetta/service/dispatch/push/dispatchers/VoIPApns.java

@@ -0,0 +1,105 @@
+package im.rosetta.service.dispatch.push.dispatchers;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.concurrent.ExecutionException;
+
+import com.eatthepath.pushy.apns.ApnsClient;
+import com.eatthepath.pushy.apns.ApnsClientBuilder;
+import com.eatthepath.pushy.apns.PushNotificationResponse;
+import com.eatthepath.pushy.apns.util.SimpleApnsPushNotification;
+import com.eatthepath.pushy.apns.util.TokenUtil;
+
+import im.rosetta.service.dispatch.push.Pusher;
+import im.rosetta.service.dispatch.runtime.PushType;
+
+public class VoIPApns extends Pusher {
+
+    private ApnsClient client;
+    private String topic;
+    private boolean isInitialized = false;
+
+    public VoIPApns(){
+        this.initializeApns();
+    }
+
+    private void initializeApns() {
+        try {
+            String p12Path = System.getenv("APNS_KEY_PATH");
+            String p12Password = System.getenv("APNS_P12_PASSWORD");
+            String bundleId = System.getenv("IOS_BUNDLE_ID");
+
+            if (p12Path == null || bundleId == null) {
+                throw new IllegalStateException("APNS_P12_PATH and IOS_BUNDLE_ID must be set");
+            }
+
+            this.topic = bundleId + ".voip";
+
+            this.client = new ApnsClientBuilder()
+                    .setApnsServer(ApnsClientBuilder.PRODUCTION_APNS_HOST)
+                    .setClientCredentials(new File(p12Path), p12Password)
+                    .build();
+            this.isInitialized = true;
+        } catch (Exception e) {
+            this.isInitialized = false;
+            e.printStackTrace();
+        }        
+    }
+
+    @Override
+    public void sendPush(String token, HashMap<String, String> data) {
+        if(!this.isInitialized) {
+            /**
+             * Нет файла для инициализации APNs клиента, пропускаем отправку VoIP уведомлений
+             */
+            return;
+        }
+        if(data.get("type") != PushType.CALL) {
+            /**
+             * Для VoIP APNs отправляем уведомления только о входящих звонках
+             */
+            return;
+        }
+        try {
+            String normalizedToken = TokenUtil.sanitizeTokenString(token);
+
+            String payload = """
+                {
+                  "aps": { "content-available": 1 },
+                  "type": "CALL",
+                  "callId": "%s",
+                  "from": "%s",
+                  "joinToken": "%s"
+                }
+                """.formatted(
+                    escape(data.getOrDefault("callId", "")),
+                    escape(data.getOrDefault("dialog", "")),
+                    escape(data.getOrDefault("joinToken", ""))
+                );
+
+            SimpleApnsPushNotification push = new SimpleApnsPushNotification(
+                    normalizedToken,
+                    topic,
+                    payload,
+                    null,          // invalidation time
+                    com.eatthepath.pushy.apns.DeliveryPriority.IMMEDIATE,
+                    com.eatthepath.pushy.apns.PushType.VOIP  // apns-push-type: voip
+            );
+
+            PushNotificationResponse<SimpleApnsPushNotification> response = client.sendNotification(push).get();
+
+            if (!response.isAccepted()) {
+                System.err.println("VoIP push rejected: " + response.getRejectionReason());
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
+        }
+    }
+
+    private String escape(String v) {
+        return v == null ? "" : v.replace("\\", "\\\\").replace("\"", "\\\"");
+    }
+
+}

src/main/java/im/rosetta/service/dispatch/runtime/PushType.java

@@ -0,0 +1,23 @@
+package im.rosetta.service.dispatch.runtime;
+
+/**
+ * Типы PUSH уведомлений, которые отправляются клиентам при получении новых сообщений, звонков и т.д.
+ */
+public class PushType {
+    /**
+     * Новое личное сообщение
+     */
+    public static final String PERSONAL_MESSAGE = "personal_message";
+    /**
+     * Новое групповое сообщение
+     */
+    public static final String GROUP_MESSAGE = "group_message";
+    /**
+    * Входящий звонок
+    */
+    public static final String CALL = "call";
+    /**
+     * Прочтение сообщения для очистки отправленных уведомлений
+     */
+    public static final String READ = "read";
+}

src/main/java/im/rosetta/service/services/DeviceService.java

@@ -1,5 +1,6 @@
 package im.rosetta.service.services;
 
+import java.util.HashSet;
 import java.util.List;
 
 import im.rosetta.client.tags.ECIAuthentificate;
@@ -82,4 +83,18 @@ public class DeviceService extends Service<DeviceRepository> {
         return device.getSyncTime();
     }
 
+    /**
+     * Получить публичные ключи пользователей, которые связаны с данным deviceId
+     * @param deviceId ID устройства
+     * @return набор публичных ключей пользователей, которые связаны с данным deviceId
+     */
+    public HashSet<String> getPublicKeysByDeviceId(String deviceId) {
+        List<Device> devices = this.getRepository().findAllByField("deviceId", deviceId);
+        HashSet<String> publicKeys = new HashSet<>();
+        for(Device device : devices) {
+            publicKeys.add(device.getPublicKey());
+        }
+        return publicKeys;
+    }
+
 }

src/main/java/im/rosetta/service/services/ForwardUnitService.java

@@ -7,6 +7,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import im.rosetta.calls.CallManager;
+import im.rosetta.calls.CallSession;
 import im.rosetta.client.ClientManager;
 import im.rosetta.logger.Logger;
 import im.rosetta.logger.enums.Color;
@@ -33,10 +35,12 @@ public class ForwardUnitService {
     private Set<SFU> sfuConnections = ConcurrentHashMap.newKeySet();
     private ClientManager clientManager;
     private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    private CallManager callManager;
 
-    public ForwardUnitService(Logger logger, ClientManager clientManager) {
+    public ForwardUnitService(Logger logger, ClientManager clientManager, CallManager callManager) {
         this.logger = logger;
         this.clientManager = clientManager;
+        this.callManager = callManager;
         this.sfuConnectionsSheduler();
     }
 
@@ -135,25 +139,33 @@ public class ForwardUnitService {
     
     public void onPeerDisconnected(DisconnectedPeer disconnectedPeer) throws ProtocolException {
         Room room = disconnectedPeer.getRoom();
-        if(disconnectedPeer.getReason() != DisconnectReason.FAILED){
+        CallSession callSession = this.callManager.getCallSession(room);
+        callSession.leaveCall(disconnectedPeer.getPeerId());
+
+        if(disconnectedPeer.getReason() == DisconnectReason.FAILED){
             /**
-             * Если у нас произошло штатное отключение, а не в результате обрыва связи - то не нужно отправлять
-             * оппонентам пакеты о том, что участник отключился в результате обрыва связи.
+             * Произошло нештатное отключение клиента от сервера SFU, например, из-за сбоя сети
              */
-            return;
-        }
-        for(String peerId : room.getParticipants()) {
-            /**
-             * Уведомляем все пиры, что соединение с пиром было потеряно
-             */
-            if(room.getParticipants().size() == 1) {
-                /**
-                 * Звонок был завершен, так как в комнате остался только один участник, который не может продолжать звонок в одиночку.
-                 */
+            if(callSession.shouldRemove()){
                 Packet26SignalPeer packet = new Packet26SignalPeer();
                 packet.setSignalType(NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED);
-                this.clientManager.sendPacketToAuthorizedPK(peerId, packet);
+                callSession.sendPacket(packet, null);
+                this.callManager.removeSession(callSession);
             }
+            return;
+        }
+        if(disconnectedPeer.getReason() == DisconnectReason.CLOSED){
+            /**
+             * Клиент намеренно покинул звонок, например, отключился от SFU сервера, так как завершил звонок или вышел из комнаты
+             * (например если клиент отрабатывает выход из звонка по кнопке END не правильно)
+             */
+            if(callSession.shouldRemove()){
+                Packet26SignalPeer packet = new Packet26SignalPeer();
+                packet.setSignalType(NetworkSignalType.END_CALL);
+                callSession.sendPacket(packet, null);
+                this.callManager.removeSession(callSession);
+            }
+            return;
         }
     }
 
@@ -255,5 +267,19 @@ public class ForwardUnitService {
         return iceServers;
     }
 
+    /**
+     * Получить комнату по ее ID, который используется на сервере SFU для идентификации комнаты.
+     * @param roomId ID комнаты на сервере SFU
+     * @return комната Room если найдена, иначе null
+     */
+    public Room getRoomById(String roomId) {
+        for(SFU sfu : this.sfuConnections) {
+            Room room = sfu.getRoom(roomId);
+            if(room != null) {
+                return room;
+            }
+        }
+        return null;
+    }
 
 }

src/main/java/im/rosetta/service/services/UserService.java

@@ -59,57 +59,4 @@ public class UserService extends Service<UserRepository> {
         User user = this.getRepository().findByField("username", username);
         return user != null;
     }
-
-    /**
-     * Подписывает пользователя на пуш уведомления, добавляя токен в его список токенов. Если токен уже был добавлен, то ничего не произойдет.
-     * @param user пользователь, которого нужно подписать на пуш уведомления
-     * @param notificationToken токен пуш уведомлений, который нужно добавить пользователю. Если токен уже был добавлен, то ничего не произойдет
-     */
-    public void subscribeToPushNotifications(User user, String notificationToken) {
-        List<String> tokens = user.getNotificationsTokens();
-        if(tokens.contains(notificationToken)){
-            return;
-        }
-        tokens.add(notificationToken);
-        user.setNotificationsTokens(tokens);
-        this.getRepository().update(user);
-    }
-
-    /**
-     * Отписывает пользователя от пуш уведомлений, удаляя токен из его списка токенов. Если токена не было, то ничего не произойдет.
-     * @param user пользователь, которого нужно отписать от пуш уведомлений
-     * @param notificationToken токен пуш уведомлений, который нужно удалить у пользователя. Если токена не было, то ничего не произойдет
-     */
-    public void unsubscribeFromPushNotifications(User user, String notificationToken) {
-        List<String> tokens = user.getNotificationsTokens();
-        if(!tokens.contains(notificationToken)){
-            return;
-        }
-        tokens.remove(notificationToken);
-        user.setNotificationsTokens(tokens);
-        this.getRepository().update(user);
-    }
-
-    /**
-     * Получает список токенов пуш уведомлений пользователя
-     * @param user пользователь, у которого нужно получить список токенов пуш уведомлений
-     * @return список токенов пуш уведомлений пользователя
-     */
-    public List<String> getNotificationsTokens(User user) {
-        return user.getNotificationsTokens();
-    }
-
-    /**
-     * Получает список токенов пуш уведомлений пользователя
-     * @param publicKey публичный ключ пользователя, у которого нужно получить список токенов пуш уведомлений
-     * @return список токенов пуш уведомлений пользователя
-     */
-    public List<String> getNotificationsTokens(String publicKey) {
-        User user = this.getRepository().findByField("publicKey", publicKey);
-        if(user == null){
-            return null;
-        }
-        return user.getNotificationsTokens();
-    }
-
 }

src/main/java/io/g365sfu/Room.java

@@ -157,4 +157,12 @@ public class Room {
         buffer.flip();
         this.sfu.getConnection().send(buffer);
     }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (obj == null || getClass() != obj.getClass()) return false;
+        Room room = (Room) obj;
+        return roomId.equals(room.roomId);
+    }
 }

src/main/java/io/orprotocol/Server.java

@@ -150,6 +150,7 @@ public class Server extends WebSocketServer {
             }
         } catch (Exception e) {
             client.disconnect(ServerFailures.BAD_PACKET);
+            e.printStackTrace();
         }
     }    
 

src/main/java/io/orprotocol/client/Client.java

@@ -239,17 +239,18 @@ public class Client {
      * @param client клиент
      * @return true если это один и тот же клиент, false если нет
      */
-    public boolean equals(Client client) {
-        if(client == null){
-            return false;
-        }
-        if(!(client instanceof Client)){
-            return false;
-        }
-        if(!client.getClientId().equals(this.clientId)){
-            return false;
-        }
-        return true;
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (obj == null || getClass() != obj.getClass()) return false;
+
+        Client client = (Client) obj;
+        return this.clientId != null && this.clientId.equals(client.clientId);
+    }
+
+    @Override
+    public int hashCode() {
+        return this.clientId == null ? 0 : this.clientId.hashCode();
     }
 
 }

src/main/resources/hibernate.cfg.xml

@@ -9,6 +9,7 @@
         <mapping class="im.rosetta.database.entity.Device"/>
         <mapping class="im.rosetta.database.entity.Group"/>
         <mapping class="im.rosetta.database.entity.Buffer"/>
+        <mapping class="im.rosetta.database.entity.PushToken"/>
         
     </session-factory>
 </hibernate-configuration>