@Override
    public <T> Mono<Long> countDistinct(Class<T> clazz, Filter filter, String groupByFieldName) {
        MongoCollection<?> collection = context.getCollection(clazz);
        List<Bson> pipeline = List.of(Aggregates.match(filter),
                Aggregates.project(Projections.fields(Projections.excludeId(),
                        Projections.include(groupByFieldName))),
                Aggregates.group("$"
                        + groupByFieldName, Accumulators.sum("count", 1)));
        AggregatePublisher<Document> count = collection.aggregate(pipeline, Document.class);
        return Mono.from(count)
                .map(document -> ((Number) document.get("count")).longValue())
                .defaultIfEmpty(0L);
    }
apply the shard key to the filter so that mongos know to which MongoDB the command needs to
be sent for better performance
turms-server-common/src/main/java/im/turms/server/common/storage/mongo/operation/TurmsMongoOperations.java