private Mono<Boolean> closeSessionsWithConflictedDeviceTypes(
            @NotNull Long userId,
            @NotNull @ValidDeviceType DeviceType deviceType,
            @NotNull UserSessionsStatus sessionsStatus) {
        try {
            Validator.notNull(userId, "userId");
            Validator.notNull(deviceType, "deviceType");
            DeviceTypeUtil.validDeviceType(deviceType);
            Validator.notNull(sessionsStatus, "sessionsStatus");
        } catch (ResponseException e) {
            return Mono.error(e);
        }
        Set<DeviceType> conflictedDeviceTypes =
                userSimultaneousLoginService.getConflictedDeviceTypes(deviceType);
        Map<String, Set<DeviceType>> nodeIdToDeviceTypes = null;
        for (DeviceType conflictedDeviceType : conflictedDeviceTypes) {
            String nodeId = sessionsStatus.getNodeIdIfActive(conflictedDeviceType);
            if (nodeId != null) {
                if (nodeIdToDeviceTypes == null) {
                    nodeIdToDeviceTypes = CollectionUtil.newMapWithExpectedSize(3);
                }
                nodeIdToDeviceTypes
                        .computeIfAbsent(nodeId, key -> CollectionUtil.newSetWithExpectedSize(3))
                        .add(deviceType);
            }
        }
        if (nodeIdToDeviceTypes == null) {
            return PublisherPool.TRUE;
        }
        Set<Map.Entry<String, Set<DeviceType>>> entries = nodeIdToDeviceTypes.entrySet();
        List<Mono<Boolean>> requests = new ArrayList<>(entries.size());
        for (Map.Entry<String, Set<DeviceType>> entry : entries) {
            String nodeId = entry.getKey();
            Set<DeviceType> deviceTypes = entry.getValue();
            SetUserOfflineRequest request = new SetUserOfflineRequest(
                    userId,
                    deviceTypes,
                    SessionCloseStatus.DISCONNECTED_BY_CLIENT);
            requests.add(node.getRpcService()
                    .requestResponse(nodeId, request)
                    .onErrorResume(ConnectionNotFound.class, t -> {
                        // The connection may not exist if a network error occurred between the
                        // local node
                        // and the target node, or the target node is dead (if it is an unknown
                        // node)
                        // without clearing its user sessions in Redis.

                        // For the first case (network problem), or we are not sure whether the
                        // target node is dead,
                        // we keep returning the expected INTERNAL_SERVER_ERROR to the client until
                        // its TTL expires.
                        if (node.getDiscoveryService()
                                .isKnownMember(nodeId)) {
                            return Mono.error(t);
                        }
                        // For the second case (dead target node), we consider the user sessions
                        // already offline,
                        // so we return true for the client to log in for better user experience.
                        return PublisherPool.TRUE;
                    }));
        }
        return PublisherUtil.areAllTrue(requests);
    }
For the case that the client recovers from UDP to TCP/WebSocket: Return the local session if
the client connects to the machine that owns the existing session; Return a new session and
disconnect the remote session if the existing session is on a different machine.
turms-gateway/src/main/java/im/turms/gateway/domain/session/service/SessionService.java