package com.github.microservice.components.data.mongo.queue.dao.impl;

import com.github.microservice.components.data.mongo.mongo.helper.DBHelper;
import com.github.microservice.components.data.mongo.queue.dao.extend.ExecQueueDaoExtend;
import com.github.microservice.components.data.mongo.queue.domain.ExecQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;

import java.util.List;

public class ExecQueueDaoImpl implements ExecQueueDaoExtend {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private DBHelper dbHelper;

    @Override
    public boolean updateStatus(String id, ExecQueue.ExecQueueStatus status) {
        Update update = new Update();
        update.set("status", status);
        this.dbHelper.updateTime(update);
        return this.mongoTemplate.updateFirst(Query.query(Criteria.where("_id").is(id)), update, ExecQueue.class).getModifiedCount() > 0;
    }

    @Override
    public List<ExecQueue> scanWait(int maxCount) {
        final long workTime = this.dbHelper.getTime();

        Query query = new Query(
                Criteria.where("status").is(ExecQueue.ExecQueueStatus.Wait)
                        .and("nextTime").lt(this.dbHelper.getTime())
                        // 尝试次数未超过最大次数
                        .and("$expr").lte(new String[]{"$currentTryCount", "$maxTryCount"})
        );
        query.with(Sort.by(Sort.Direction.ASC, "nextTime", "workTime"));

        Update update = new Update();
        update.set("status", ExecQueue.ExecQueueStatus.Work);
        update.set("workTime", workTime);

        final BulkOperations bulkOperations = this.mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, ExecQueue.class);
        for (int i = 0; i < maxCount; i++) {
            bulkOperations.updateOne(query, update);
        }
        bulkOperations.execute();

        return this.mongoTemplate.find(Query.query(Criteria.where("workTime").is(workTime)), ExecQueue.class);
    }

    @Override
    public long restoreWaitStatus(long executeTimeOut) {
        Query query = new Query(
                Criteria.where("status").is(ExecQueue.ExecQueueStatus.Work)
                        .and("workTime").lt(this.dbHelper.getTime() - executeTimeOut)
        );

        Update update = new Update();
        update.set("status", ExecQueue.ExecQueueStatus.Wait);
        update.inc("currentTryCount", 1);
        this.dbHelper.updateTime(update);
        return this.mongoTemplate.updateMulti(query, update, ExecQueue.class).getModifiedCount();
    }

}
