package tech.easyflow.agent.support;

import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import tech.easyflow.common.cache.RedisLockExecutor;

import java.math.BigInteger;
import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

/**
 * 在同一 Agent 维度串行执行绑定变更，并将锁持有到数据库事务结束。
 */
@Component
public class AgentBindingLockExecutor {

    private static final Logger LOG = LoggerFactory.getLogger(AgentBindingLockExecutor.class);
    private static final String LOCK_KEY_PREFIX = "easyflow:lock:agent:binding:";
    private static final Duration LOCK_WAIT_TIMEOUT = Duration.ofSeconds(2);
    private static final Duration LOCK_LEASE_TIMEOUT = Duration.ofSeconds(30);

    private final RedisLockExecutor redisLockExecutor;
    private final ScheduledExecutorService renewExecutor;

    /**
     * 创建 Agent 绑定锁执行器。
     *
     * @param redisLockExecutor Redis 分布式锁执行器
     */
    @Autowired
    public AgentBindingLockExecutor(RedisLockExecutor redisLockExecutor) {
        this(
                redisLockExecutor,
                Executors.newSingleThreadScheduledExecutor(new AgentBindingLockRenewThreadFactory())
        );
    }

    /**
     * 创建使用指定续期执行器的 Agent 绑定锁执行器。
     *
     * @param redisLockExecutor Redis 分布式锁执行器
     * @param renewExecutor 锁续期执行器
     */
    AgentBindingLockExecutor(RedisLockExecutor redisLockExecutor,
                             ScheduledExecutorService renewExecutor) {
        this.redisLockExecutor = redisLockExecutor;
        this.renewExecutor = renewExecutor;
    }

    /**
     * 在 Agent 绑定锁内执行任务。
     *
     * <p>存在活动事务时，锁会在事务完成后释放，避免提交前出现并发写入窗口。</p>
     *
     * @param agentId Agent ID
     * @param task 绑定变更任务
     * @param <T> 返回类型
     * @return 任务结果
     */
    public <T> T execute(BigInteger agentId, Supplier<T> task) {
        if (agentId == null) {
            throw new IllegalArgumentException("agentId 不能为空");
        }
        String lockKey = LOCK_KEY_PREFIX + agentId;
        if (TransactionSynchronizationManager.hasResource(lockKey)) {
            return task.get();
        }
        RedisLockExecutor.LockHandle lockHandle = redisLockExecutor.acquire(
                lockKey,
                LOCK_WAIT_TIMEOUT,
                LOCK_LEASE_TIMEOUT
        );
        AtomicBoolean leaseValid = new AtomicBoolean(true);
        ScheduledFuture<?> renewTask = scheduleRenew(agentId, lockHandle, leaseValid);
        Runnable releaseAction = () -> {
            renewTask.cancel(false);
            lockHandle.release();
        };
        boolean releaseAfterTransaction = false;
        try {
            releaseAfterTransaction =
                    registerTransactionRelease(lockKey, releaseAction, leaseValid);
            T result = task.get();
            if (!leaseValid.get()) {
                throw new IllegalStateException("Agent 绑定锁已失效，当前操作已取消");
            }
            return result;
        } finally {
            if (!releaseAfterTransaction) {
                releaseAction.run();
            }
        }
    }

    /**
     * 注册事务完成后的锁释放动作。
     *
     * @param lockKey 锁键
     * @param releaseAction 锁释放动作
     * @param leaseValid 锁租期有效标记
     * @return 已注册事务回调时返回 {@code true}
     */
    private boolean registerTransactionRelease(String lockKey,
                                               Runnable releaseAction,
                                               AtomicBoolean leaseValid) {
        if (!TransactionSynchronizationManager.isActualTransactionActive()
                || !TransactionSynchronizationManager.isSynchronizationActive()) {
            return false;
        }
        TransactionSynchronizationManager.bindResource(lockKey, Boolean.TRUE);
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void beforeCommit(boolean readOnly) {
                if (!leaseValid.get()) {
                    throw new IllegalStateException("Agent 绑定锁已失效，事务禁止提交");
                }
            }

            @Override
            public void afterCompletion(int status) {
                try {
                    TransactionSynchronizationManager.unbindResourceIfPossible(lockKey);
                } finally {
                    releaseAction.run();
                }
            }
        });
        return true;
    }

    /**
     * 定期续期 Agent 绑定锁，保证长事务中的早期锁不会在提交前过期。
     *
     * @param agentId Agent ID
     * @param lockHandle 锁句柄
     * @param leaseValid 锁租期有效标记
     * @return 续期任务
     */
    private ScheduledFuture<?> scheduleRenew(BigInteger agentId,
                                             RedisLockExecutor.LockHandle lockHandle,
                                             AtomicBoolean leaseValid) {
        long renewIntervalMillis = Math.max(LOCK_LEASE_TIMEOUT.toMillis() / 3L, 1000L);
        return renewExecutor.scheduleWithFixedDelay(() -> {
            try {
                if (lockHandle.renew()) {
                    return;
                }
                leaseValid.set(false);
                LOG.warn("Agent 绑定锁续期失败，agentId={}", agentId);
            } catch (RuntimeException exception) {
                leaseValid.set(false);
                LOG.error("Agent 绑定锁续期异常，agentId={}", agentId, exception);
            }
        }, renewIntervalMillis, renewIntervalMillis, TimeUnit.MILLISECONDS);
    }

    /**
     * 关闭锁续期线程。
     */
    @PreDestroy
    public void destroy() {
        renewExecutor.shutdownNow();
    }

    /**
     * Agent 绑定锁续期线程工厂。
     */
    private static final class AgentBindingLockRenewThreadFactory implements ThreadFactory {

        private final AtomicInteger index = new AtomicInteger(1);

        /**
         * 创建守护续期线程。
         *
         * @param runnable 续期任务
         * @return 续期线程
         */
        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable);
            thread.setName("agent-binding-lock-renew-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        }
    }
}
