private Mono<Set<Long>> forwardClientMessageToNodes(
            ByteBuf messageData,
            Map<String, Set<Long>> nodeIdToRecipientIds,
            Set<UserSessionId> excludedUserSessionIds) {
        Set<String> nodeIds = nodeIdToRecipientIds.keySet();
        int nodeIdCount = nodeIds.size();
        if (nodeIdCount == 0) {
            messageData.release();
            return PublisherPool.emptySet();
        }
        if (nodeIdCount == 1) {
            String nodeId = nodeIds.iterator()
                    .next();
            return forwardClientMessageToNode(messageData,
                    nodeId,
                    nodeIdToRecipientIds.get(nodeId),
                    excludedUserSessionIds,
                    null);
        }
        List<Mono<Set<Long>>> forwardClientMessageMonos = new ArrayList<>(nodeIdCount);
        messageData.retain(nodeIdCount);
        for (Map.Entry<String, Set<Long>> nodeIdAndRecipientIds : nodeIdToRecipientIds.entrySet()) {
            forwardClientMessageMonos.add(forwardClientMessageToNode(messageData,
                    nodeIdAndRecipientIds.getKey(),
                    nodeIdAndRecipientIds.getValue(),
                    excludedUserSessionIds,
                    null));
        }
        return collectOfflineRecipientIds(forwardClientMessageMonos)
                .doFinally(signal -> messageData.release());
    }
@return offline user IDs
@implNote Return {@code Set<Long>} instead of {@code Long} for consistency.
turms-server-common/src/main/java/im/turms/server/common/infra/message/OutboundMessageManager.java