@Override
    public Mono<Void> ensureIndexesAndShards(
            Collection<Class<?>> classes,
            @Nullable BiPredicate<Class<?>, CompoundIndex> customCompoundIndexFilter,
            @Nullable BiPredicate<Class<?>, Field> customIndexFilter) {
        Mono<Void> ensureIndexes = Mono.empty();
        for (Class<?> clazz : classes) {
            MongoEntity<?> entity = context.getEntity(clazz);
            List<Index> indexes = entity.indexes();
            List<IndexModel> indexModels = new ArrayList<>(indexes.size());
            for (im.turms.server.common.storage.mongo.entity.CompoundIndex compoundIndex : entity
                    .compoundIndexes()) {
                if (customCompoundIndexFilter == null) {
                    indexModels.add(compoundIndex.model());
                } else {
                    boolean isOptional = compoundIndex.index()
                            .ifExist().length > 0
                            || compoundIndex.index()
                                    .ifNotExist().length > 0;
                    if (isOptional) {
                        if (customCompoundIndexFilter.test(clazz, compoundIndex.index())) {
                            indexModels.add(compoundIndex.model());
                        }
                    } else {
                        indexModels.add(compoundIndex.model());
                    }
                }
            }
            for (Index index : indexes) {
                if (index.indexed()
                        .optional()) {
                    if (customIndexFilter != null && customIndexFilter.test(clazz, index.field())) {
                        indexModels.add(index.model());
                    }
                } else {
                    indexModels.add(index.model());
                }
            }
            if (!indexModels.isEmpty()) {
                ensureIndexes = ensureIndexes
                        .then(Mono.defer(() -> ensureIndexes(entity.entityClass(), indexModels)));
            }
        }
        return ensureIndexes.then(enableSharding())
                .then(Mono.defer(() -> {
                    Mono<Void> shardEntities = Mono.empty();
                    for (Class<?> clazz : classes) {
                        MongoEntity<?> entity = context.getEntity(clazz);
                        shardEntities = shardEntities.then(Mono.defer(() -> shard(entity)));
                    }
                    return shardEntities;
                }));
    }
@implNote Send requests to ensure indexes and shard collections synchronously to avoid mongo
          servers throwing "LockBusy" error (error code 46)
turms-server-common/src/main/java/im/turms/server/common/storage/mongo/operation/TurmsMongoOperations.java