package jnpf.flowable.service;

import com.baomidou.dynamic.datasource.tx.*;
import jnpf.exception.WorkFlowException;
import jnpf.flowable.mapper.DmsLifecycleEngineRecoveryMapper;
import jnpf.flowable.model.flowable.FlowAbleUrl;
import lombok.Data;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.SqlSessionUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.function.Supplier;

/** Durable DMS-only recovery; native business CRUD and ordinary engine calls are unchanged. */
@Service
public class DmsLifecycleEngineRecovery {
    private static final Logger LOG=LoggerFactory.getLogger(DmsLifecycleEngineRecovery.class);
    public static final String CORRELATION_VARIABLE="dmsLifecycleRecoveryId";
    private final SqlSessionTemplate sessions;
    private final FlowAbleUrl engine;
    private final boolean enabled;
    public DmsLifecycleEngineRecovery(SqlSessionTemplate sessions,FlowAbleUrl engine,
            @Value("${dms.business-occupancy.enabled:false}") boolean enabled) {
        this.sessions=sessions;this.engine=engine;this.enabled=enabled;
    }
    private DmsLifecycleEngineRecoveryMapper mapper() {return sessions.getMapper(DmsLifecycleEngineRecoveryMapper.class);}

    /** A log is committed BEFORE starting the engine; the native transaction holds its recovery gate. */
    public Started start(String tenant,String task,String deployment,Map<String,Object> variables) throws WorkFlowException {
        if(!enabled || task==null || task.isBlank()) throw new IllegalStateException("DMS 引擎恢复未启用或任务标识缺失");
        if(variables.containsKey(CORRELATION_VARIABLE)) throw new IllegalArgumentException("DMS 引擎恢复变量冲突");
        var row=new Record();row.setId(UUID.randomUUID().toString());row.setTenantId(tenant==null || tenant.isBlank()?"0":tenant);row.setTaskId(task);
        holdNativeGate(row.getId());
        independentChecked(()->{if(mapper().insert(row)!=1)throw new IllegalStateException("DMS 引擎恢复记录未保存");return null;});
        var actual=new HashMap<>(variables);actual.put(CORRELATION_VARIABLE,row.getId());
        String instance=engine.startInstance(deployment,actual);
        if(instance==null || instance.isBlank()) throw new WorkFlowException("DMS 引擎启动结果不明确，需按恢复记录核对");
        row.setInstanceId(instance);
        var ticket=DmsLifecycleEngineCompensation.register(instance,()->{
            independentChecked(()->{
                // Even a commit-response failure must not delete an instance already referenced by committed data.
                if(mapper().committed(row)) {mapper().finish(row.getId(),"KEPT");return null;}
                try {engine.deleteDmsInstanceStrict(instance,reason(row.getId()));}
                catch(WorkFlowException failure) {throw new IllegalStateException(failure);}
                mapper().finish(row.getId(),"DELETED");return null;
            });
        });
        try {
            independentChecked(()->{if(mapper().attach(row.getId(),instance)!=1)throw new IllegalStateException("DMS 引擎实例恢复关联未保存");return null;});
        } catch(RuntimeException failure) {ticket.compensate(failure);throw failure;}
        return new Started(instance,ticket);
    }
    private void holdNativeGate(String id) {
        withVerifiedConnection(()->{
            if(!mapper().tryLock(id)) throw new IllegalStateException("DMS 引擎恢复门闩冲突");
            return null;
        });
    }
    private <T> T independentChecked(Supplier<T> action) {
        return independent(()->withVerifiedConnection(action));
    }
    private <T> T withVerifiedConnection(Supplier<T> action) {
        String xid=TransactionContext.getXID();
        if(xid==null || TransactionSynchronizationManager.isActualTransactionActive())
            throw new IllegalStateException("DMS 引擎启动必须属于原生动态事务");
        var session=SqlSessionUtils.getSqlSession(sessions.getSqlSessionFactory(),sessions.getExecutorType(),sessions.getPersistenceExceptionTranslator());
        try {
            var connection=session.getConnection();
            if(!(connection instanceof ConnectionProxy proxy) || ConnectionFactory.getConnection(xid,proxy.getDs())!=connection
                    || connection.getAutoCommit() || connection.isReadOnly()
                    || connection.getTransactionIsolation()!=java.sql.Connection.TRANSACTION_READ_COMMITTED)
                throw new IllegalStateException("DMS 引擎恢复门闩未绑定原生事务连接");
            return action.get();
        } catch(java.sql.SQLException failure) {throw new IllegalStateException(failure);}
        finally {SqlSessionUtils.closeSqlSession(session,sessions.getSqlSessionFactory());}
    }
    /** REQUIRES_NEW suspends the actual dynamic transaction; Spring propagation is not substituted. */
    static <T> T independent(Supplier<T> action) {
        if(TransactionSynchronizationManager.isActualTransactionActive()) throw new IllegalStateException("不能混用DMS恢复事务管理器");
        // Version 4.3.1 suspends only XID, not its thread-local synchronization list.
        // Isolate callbacks here so an independent journal transaction cannot consume native callbacks.
        String outerXid=TransactionContext.getXID();
        var outerCallbacks=TransactionContext.getSynchronizations();
        TransactionContext.removeSynchronizations();
        var info=new TransactionalInfo();info.setPropagation(DsPropagation.REQUIRES_NEW);
        info.setRollbackFor(new Class[]{Exception.class});info.setNoRollbackFor(new Class[]{});
        try {
            @SuppressWarnings("unchecked") T result=(T)new TransactionalTemplate().execute(new TransactionalExecutor() {
                public Object execute() {return action.get();}
                public TransactionalInfo getTransactionInfo() {return info;}
            });return result;
        } catch(RuntimeException | Error failure) {throw failure;}
        catch(Throwable failure) {throw new IllegalStateException("DMS 独立恢复事务失败",failure);}
        finally {
            TransactionContext.removeSynchronizations();
            // afterCompletion runs after LocalTxUtil has already removed XID.
            // Its callback snapshot is still being iterated, but no live transaction can be re-registered.
            if(outerXid!=null && outerXid.equals(TransactionContext.getXID()))
                outerCallbacks.forEach(TransactionContext::registerSynchronization);
        }
    }
    @Scheduled(cron="${dms.business-occupancy.engine-recovery-cron:0 * * * * ?}")
    public void recoverPending() {
        if(!enabled) return;
        for(var candidate:mapper().pending(100)) {
            try {recover(candidate.getId());}
            catch(RuntimeException failure) {LOG.error("DMS engine recovery scan failed, recoveryId={}",candidate.getId(),failure);}
        }
    }
    public void recover(String id) {
        independentChecked(()->{
            if(!mapper().tryLock(id)) return null; // A live native transaction still owns this instance.
            var row=mapper().get(id);
            if(row==null || !"PENDING".equals(row.getStatus())) return null;
            if(row.getInstanceId()==null || row.getInstanceId().isBlank()) {
                mapper().finish(id,"NEEDS_REVIEW");return null; // Never guess which remote instance to delete.
            }
            if(mapper().committed(row)) {mapper().finish(id,"KEPT");return null;}
            try {
                engine.deleteDmsInstanceStrict(row.getInstanceId(),reason(id));
                mapper().finish(id,"DELETED");
            } catch(Exception failure) {
                String type=failure.getClass().getName();
                mapper().failed(id,type.substring(0,Math.min(200,type.length())));
                LOG.warn("DMS engine deletion unconfirmed, recoveryId={}",id);
            }
            return null;
        });
    }
    private static String reason(String id) {return "dmsLifecycleRollback:"+id;}
    public record Started(String instanceId,DmsLifecycleEngineCompensation compensation) { }
    @Data public static class Record {
        private String id;
        private String tenantId;
        private String taskId;
        private String instanceId;
        private String status;
    }
}