public Mono<Collection<UserSessionsInfo>> queryUserSessions(Set<Long> userIds) {
        int userCount = userIds.size();
        if (userCount == 0) {
            return PublisherPool.emptyCollection();
        }
        List<Mono<UserSessionsStatus>> sessionStatusMonos = new ArrayList<>(userCount);
        for (Long userId : userIds) {
            sessionStatusMonos.add(userStatusService.getUserSessionsStatus(userId));
        }
        return Flux.merge(sessionStatusMonos)
                .collect(CollectorUtil.toList(userCount))
                .flatMap(sessionsStatuses -> {
                    // Find which nodes the users are in
                    List<UserSessionsInfo> offlineUserSessions = null;
                    Map<String, Set<Long>> nodeIdToUserIds = new HashMap<>(16);
                    for (UserSessionsStatus sessionsStatus : sessionsStatuses) {
                        Map<DeviceType, UserDeviceSessionInfo> deviceTypeToSessionInfos =
                                sessionsStatus.getDeviceTypeToSessionInfo();
                        if (deviceTypeToSessionInfos.isEmpty()) {
                            if (offlineUserSessions == null) {
                                offlineUserSessions = new ArrayList<>(userCount);
                            }
                            offlineUserSessions.add(new UserSessionsInfo(
                                    sessionsStatus.getUserId(),
                                    UserStatus.OFFLINE,
                                    Collections.emptyList()));
                        } else {
                            for (UserDeviceSessionInfo sessionInfo : deviceTypeToSessionInfos
                                    .values()) {
                                if (sessionInfo.isActive()) {
                                    nodeIdToUserIds
                                            .computeIfAbsent(sessionInfo.getNodeId(),
                                                    key -> CollectionUtil
                                                            .newSetWithExpectedSize(userCount))
                                            .add(sessionsStatus.getUserId());
                                }
                            }
                        }
                    }
                    int nodeIdCount = nodeIdToUserIds.size();
                    if (nodeIdCount == 0) {
                        return offlineUserSessions == null
                                ? PublisherPool.emptyList()
                                : Mono.just(offlineUserSessions);
                    }
                    List<Mono<List<UserSessionsInfo>>> querySessionsRequests =
                            new ArrayList<>(nodeIdCount);
                    // Send RPC to the nodes to query sessions
                    for (Map.Entry<String, Set<Long>> nodeIdAndUserIds : nodeIdToUserIds
                            .entrySet()) {
                        querySessionsRequests
                                .add(rpcService.requestResponse(nodeIdAndUserIds.getKey(),
                                        new QueryUserSessionsRequest(nodeIdAndUserIds.getValue())));
                    }
                    List<UserSessionsInfo> finalOfflineUserSessions = offlineUserSessions;
                    return Flux.merge(querySessionsRequests)
                            .collect(CollectorUtil.toList(nodeIdCount))
                            .map(sessionsFromNodes -> mergeUserSessions(sessionsFromNodes,
                                    finalOfflineUserSessions,
                                    userCount));
                });
    }
@return includes the sessions with the {@link UserStatus#OFFLINE} status for offline users
turms-service/src/main/java/im/turms/service/domain/user/service/onlineuser/SessionService.java