package cn.axzo.workflow.core.engine.job;

import cn.axzo.workflow.common.exception.WorkflowEngineException;
import cn.axzo.workflow.core.common.utils.SpringContextUtils;
import cn.axzo.workflow.core.repository.entity.ExtAxProperty;
import cn.axzo.workflow.core.service.ExtAxPropertyService;
import com.alibaba.nacos.common.utils.ThreadUtils;
import org.apache.commons.lang.StringUtils;
import org.flowable.common.engine.impl.interceptor.CommandContext;
import org.flowable.job.service.JobHandler;
import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DuplicateKeyException;

import java.util.concurrent.TimeUnit;

import static cn.axzo.workflow.common.code.OtherRespCode.ASYNC_JOB_EXECUTION_ERROR;

public abstract class AbstractExecuteWithLockJobHandler extends AbstractJobHandler implements JobHandler {

    private static final Logger log = LoggerFactory.getLogger(AbstractExecuteWithLockJobHandler.class);

    @Override
    public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
        String processInstanceId = job.getProcessInstanceId(), jobId = job.getId();
        if (StringUtils.isBlank(processInstanceId) || StringUtils.isBlank(jobId)) {
            log.warn("processInstanceId or lockOwner is empty,cannot execute with lock,jobId:{},processInstanceId:{}", job.getId(), job.getProcessInstanceId());
            executeInternal(job, configuration, variableScope, commandContext);
            return;
        }
        try {
            if (acquireLock(processInstanceId, jobId)) {
                log.info("job acquire lock success,processInstanceId:{},jobId:{}", processInstanceId, jobId);
                executeInternal(job, configuration, variableScope, commandContext);
            } else {
                log.warn("get lock failed,processInstanceId:{},jobId:{}", processInstanceId, jobId);
                throw new WorkflowEngineException(ASYNC_JOB_EXECUTION_ERROR, processInstanceId);
            }
        } finally {
            releaseLock(processInstanceId, jobId);
        }
    }

    private boolean acquireLock(String processInstanceId, String jobId) {
        int attemptNo = 3, acquireCount = 0;
        ExtAxPropertyService extAxPropertyService = SpringContextUtils.getBean(ExtAxPropertyService.class);
        do {
            try {
                ExtAxProperty extAxProperty = new ExtAxProperty();
                extAxProperty.setName(processInstanceId);
                extAxProperty.setValue(jobId);
                extAxPropertyService.add(extAxProperty);
                return true;
            } catch (DuplicateKeyException e) {
                //删除15分钟前的数据,相当于超时时间是15分钟
                extAxPropertyService.deleteByNameWithDuration(processInstanceId, 15L, TimeUnit.MINUTES);
                //睡眠2秒
                ThreadUtils.sleep(2 * 1000);
                acquireCount++;
                log.warn("acquireLock error,processInstanceId:{},jobId:{},acquireCount:{}", processInstanceId, jobId, acquireCount);
            }
        } while (acquireCount < attemptNo);
        return false;
    }

    private void releaseLock(String processInstanceId, String jobId) {
        ExtAxPropertyService extAxPropertyService = SpringContextUtils.getBean(ExtAxPropertyService.class);
        extAxPropertyService.delete(processInstanceId, jobId);
    }

    public abstract void executeInternal(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext);
}
