@Override
    public <T> Mono<UpdateResult> upsert(@Nullable ClientSession session, T record) {
        Class<T> clazz = (Class<T>) record.getClass();
        MongoEntity<T> entity = (MongoEntity<T>) context.getEntity(record.getClass());
        MongoCollection<T> collection = context.getCollection(clazz);
        BsonDocument recordDoc = encodeEntityForUpsert(record);
        ShardKey shardKey = entity.shardKey();
        BsonDocument filter =
                new BsonDocument(DomainFieldName.ID, recordDoc.get(DomainFieldName.ID));
        if (shardKey != null) {
            applyShardKey(filter, shardKey, recordDoc);
        }
        // Remove "_id", or MongoDB will throw "Performing an update on the path '_id' would modify
        // the immutable field '_id'"
        recordDoc.remove(DomainFieldName.ID);
        BsonDocument update = new BsonDocument("$set", recordDoc);
        Publisher<UpdateResult> result = session == null
                ? collection.updateOne(filter, update, DEFAULT_UPSERT_OPTIONS)
                : collection.updateOne(session, filter, update, DEFAULT_UPSERT_OPTIONS);
        return Mono.from(result)
                .onErrorMap(MongoExceptionUtil::translate);
    }
@author James Chen
@implNote 1. We operate in an unsafe way, which means we don't check whether arguments are legal
          or not and so on 2. The publishers of mongo-java-driver are cold and the publishers of
          TurmsMongoOperations are also cold Reference:
          https://github.com/mongodb/mongo/blob/master/src/mongo/shell/utils_sh.js
turms-server-common/src/main/java/im/turms/server/common/storage/mongo/operation/TurmsMongoOperations.java