public Mono<UserSession> tryRegisterOnlineUser(
            int version,
            @NotNull Set<TurmsRequest.KindCase> permissions,
            @NotNull ByteArrayWrapper ip,
            @NotNull Long userId,
            @NotNull DeviceType deviceType,
            @Nullable Map<String, String> deviceDetails,
            @Nullable UserStatus userStatus,
            @Nullable Location location) {
        try {
            Validator.notNull(ip, "ip");
            Validator.notNull(deviceType, "deviceType");
            DeviceTypeUtil.validDeviceType(deviceType);
            Validator.notEquals(userStatus,
                    UserStatus.UNRECOGNIZED,
                    "The user status must not be UNRECOGNIZED");
            Validator.notEquals(userStatus,
                    UserStatus.OFFLINE,
                    "The user status must not be OFFLINE");
            if (location != null) {
                Validator.inRange(location.longitude(),
                        "longitude",
                        Location.LONGITUDE_MIN,
                        Location.LONGITUDE_MAX);
                Validator.inRange(location.latitude(),
                        "latitude",
                        Location.LATITUDE_MIN,
                        Location.LATITUDE_MAX);
            }
        } catch (ResponseException e) {
            return Mono.error(e);
        }
        // Must fetch the latest status instead of the status in the cache
        return userStatusService.fetchUserSessionsStatus(userId)
                .flatMap(sessionsStatus -> {
                    // 1. Close local sessions if they are already registered on Redis by other
                    // nodes to handle the following edge cases to avoid bugs when the session
                    // info in the local node is inconsistent with the one on Redis.

                    // Cases: The session exists in the local node, but:
                    // 1. Though the local node works well, Redis crashes and restart,
                    // so all session info was lost on Redis, but sessions still exist indeed.
                    // 2. The local node lost the connection to Redis, which will cause
                    // the local node to fail to refresh the heartbeat info of users on Redis.
                    UserSessionsManager manager = userIdToSessionsManager.get(userId);
                    if (manager != null) {
                        Map<DeviceType, UserDeviceSessionInfo> deviceTypeToSessionInfo =
                                sessionsStatus.getDeviceTypeToSessionInfo();
                        for (DeviceType type : manager.getLoggedInDeviceTypes()) {
                            UserDeviceSessionInfo sessionInfo = deviceTypeToSessionInfo.get(type);
                            if (sessionInfo != null
                                    && sessionInfo.isActive()
                                    && !node.getLocalMemberId()
                                            .equals(sessionInfo.getNodeId())) {
                                closeLocalSession(userId,
                                        type,
                                        SessionCloseStatus.DISCONNECTED_BY_OTHER_DEVICE)
                                        .subscribe(null,
                                                t -> LOGGER.error(
                                                        "Caught an error while closing the user session with the user ID: "
                                                                + userId,
                                                        t));
                            }
                        }
                    }
                    // If the user is offline, register the current session.
                    UserStatus existingUserStatus = sessionsStatus.getUserStatus();
                    if (existingUserStatus == UserStatus.OFFLINE) {
                        return addOnlineDeviceIfAbsent(version,
                                permissions,
                                ip,
                                userId,
                                deviceType,
                                deviceDetails,
                                userStatus,
                                location,
                                null,
                                null);
                    }
                    // If the user is already online, check if there is any device conflict.
                    UserDeviceSessionInfo sessionInfo = sessionsStatus.getDeviceTypeToSessionInfo()
                            .get(deviceType);
                    if (sessionInfo != null && sessionInfo.isActive()) {
                        UserSession session = getLocalUserSession(userId, deviceType);
                        boolean isClosedSessionOnLocal = session != null
                                && session.getConnection() != null
                                && !session.getConnection()
                                        .isConnected();
                        if (isClosedSessionOnLocal) {
                            // Note that the downstream should replace the disconnected connection
                            // with the connected TCP/WebSocket connection
                            Mono<Void> updateSessionInfoMono =
                                    userStatus == null || existingUserStatus == userStatus
                                            ? Mono.empty()
                                            : userStatusService
                                                    .updateOnlineUserStatusIfPresent(userId,
                                                            userStatus)
                                                    .then()
                                                    .onErrorComplete(t -> {
                                                        LOGGER.error(
                                                                "Failed to update the online status of the user: "
                                                                        + userId,
                                                                t);
                                                        return true;
                                                    });
                            if (location != null) {
                                updateSessionInfoMono = updateSessionInfoMono
                                        .flatMap(unused -> sessionLocationService
                                                .upsertUserLocation(userId,
                                                        deviceType,
                                                        new Date(),
                                                        location.longitude(),
                                                        location.latitude())
                                                .onErrorComplete(t -> {
                                                    LOGGER.error(
                                                            "Failed to upsert the location of the user session: {user={}, device={}}",
                                                            session.getUserId(),
                                                            session.getDeviceType(),
                                                            t);
                                                    return true;
                                                }));
                            }
                            return updateSessionInfoMono.thenReturn(session);
                        } else if (userSimultaneousLoginService
                                .shouldDisconnectLoggingInDeviceIfConflicts()) {
                            return Mono.error(ResponseException.get(
                                    ResponseStatusCode.SESSION_SIMULTANEOUS_CONFLICTS_DECLINE));
                        }
                    }
                    String expectedNodeId;
                    Long expectedDeviceTimestamp;
                    if (sessionInfo == null) {
                        expectedNodeId = null;
                        expectedDeviceTimestamp = null;
                    } else {
                        expectedNodeId = sessionInfo.getNodeId();
                        expectedDeviceTimestamp = sessionInfo.getHeartbeatTimestampSeconds();
                    }
                    return closeSessionsWithConflictedDeviceTypes(userId,
                            deviceType,
                            sessionsStatus).flatMap(
                                    wasSuccessful -> wasSuccessful
                                            ? addOnlineDeviceIfAbsent(version,
                                                    permissions,
                                                    ip,
                                                    userId,
                                                    deviceType,
                                                    deviceDetails,
                                                    userStatus,
                                                    location,
                                                    expectedNodeId,
                                                    expectedDeviceTimestamp)
                                            : Mono.error(ResponseException.get(
                                                    ResponseStatusCode.SESSION_SIMULTANEOUS_CONFLICTS_DECLINE)));
                });
    }
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