package com.easyagents.flow.core.util;

import com.easyagents.flow.core.chain.runtime.RetryableTriggerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.time.Duration;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

/**
 * 为工作流中的阻塞 I/O 提供有界并发、单目标隔离和可观测计数。
 */
public final class IoBulkhead {

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

    private static final String GLOBAL_CONCURRENCY_PROPERTY = "tinyflow.io.max-concurrency";
    private static final String TARGET_CONCURRENCY_PROPERTY = "tinyflow.io.per-target-max-concurrency";
    private static final String ACQUIRE_TIMEOUT_MILLIS_PROPERTY = "tinyflow.io.acquire-timeout-ms";
    private static final int DEFAULT_GLOBAL_CONCURRENCY = 64;
    private static final int DEFAULT_TARGET_CONCURRENCY = 16;
    private static final long DEFAULT_ACQUIRE_TIMEOUT_MILLIS = 1_000L;
    private static final int DEFAULT_MAX_TRACKED_TARGETS = 1_024;
    private static final long REJECTION_LOG_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(10);
    private static volatile IoBulkhead SHARED = new IoBulkhead(
            positiveIntProperty(GLOBAL_CONCURRENCY_PROPERTY, DEFAULT_GLOBAL_CONCURRENCY),
            positiveIntProperty(TARGET_CONCURRENCY_PROPERTY, DEFAULT_TARGET_CONCURRENCY),
            Duration.ofMillis(positiveLongProperty(
                    ACQUIRE_TIMEOUT_MILLIS_PROPERTY,
                    DEFAULT_ACQUIRE_TIMEOUT_MILLIS)),
            DEFAULT_MAX_TRACKED_TARGETS);
    private static volatile IoBulkhead DATASET = lane(
            "tinyflow.io.dataset", 32, 8, 1_000L, 512);
    private static volatile IoBulkhead STORAGE = lane(
            "tinyflow.io.storage", 24, 12, 2_000L, 256);
    private static volatile IoBulkhead DOCUMENT_PARSE = lane(
            "tinyflow.io.document-parse", 8, 4, 2_000L, 128);
    private static volatile IoBulkhead RESPONSE_AGGREGATION = lane(
            "tinyflow.io.response-aggregation", 8, 4, 2_000L, 1_024);

    private final Semaphore globalSemaphore;
    private final int perTargetConcurrency;
    private final long acquireTimeoutNanos;
    private final int maxTrackedTargets;
    private final Map<String, Semaphore> targetSemaphores = new ConcurrentHashMap<>();
    private final Object targetRegistryLock = new Object();
    private final Semaphore overflowTargetSemaphore;
    private final LongAdder acquiredCount = new LongAdder();
    private final LongAdder rejectedCount = new LongAdder();
    private final LongAdder inFlightCount = new LongAdder();
    private final LongAdder totalWaitNanos = new LongAdder();
    private final AtomicLong lastRejectionLogNanos = new AtomicLong();

    /**
     * 创建 I/O 隔离器。
     *
     * @param globalConcurrency 当前进程允许的 I/O 总并发
     * @param perTargetConcurrency 单个目标允许的 I/O 并发
     * @param acquireTimeout 等待许可的最长时间
     * @param maxTrackedTargets 最多单独跟踪的目标数量
     * @throws IllegalArgumentException 参数不合法时抛出
     */
    public IoBulkhead(
            int globalConcurrency,
            int perTargetConcurrency,
            Duration acquireTimeout,
            int maxTrackedTargets) {
        if (globalConcurrency <= 0
                || perTargetConcurrency <= 0
                || acquireTimeout == null
                || acquireTimeout.isNegative()
                || maxTrackedTargets <= 0) {
            throw new IllegalArgumentException("I/O bulkhead configuration must be positive");
        }
        this.globalSemaphore = new Semaphore(globalConcurrency, true);
        this.perTargetConcurrency = perTargetConcurrency;
        this.acquireTimeoutNanos = acquireTimeout.toNanos();
        this.maxTrackedTargets = maxTrackedTargets;
        this.overflowTargetSemaphore = new Semaphore(perTargetConcurrency, true);
    }

    /**
     * 获取进程内共享 I/O 隔离器。
     *
     * @return 共享隔离器
     */
    public static IoBulkhead shared() {
        return SHARED;
    }

    /**
     * 获取数据集读写独立隔离器。
     *
     * @return 数据集隔离器
     */
    public static IoBulkhead dataset() {
        return DATASET;
    }

    /**
     * 获取对象存储读写独立隔离器。
     *
     * @return 对象存储隔离器
     */
    public static IoBulkhead storage() {
        return STORAGE;
    }

    /**
     * 获取文档解析独立隔离器。
     *
     * @return 文档解析隔离器
     */
    public static IoBulkhead documentParse() {
        return DOCUMENT_PARSE;
    }

    /**
     * 获取必须完整物化响应的独立小并发隔离器。
     *
     * @return 响应聚合隔离器
     */
    public static IoBulkhead responseAggregation() {
        return RESPONSE_AGGREGATION;
    }

    /**
     * 原子替换全部工作流 I/O 隔离器配置。
     *
     * <p>应用应在开始处理工作流前调用。既有许可继续由旧实例释放，
     * 后续请求读取新实例，不会中断正在执行的 I/O。</p>
     *
     * @param http HTTP 请求配置
     * @param dataset 数据集配置
     * @param storage 对象存储配置
     * @param documentParse 文档解析配置
     * @param responseAggregation 响应聚合配置
     */
    public static synchronized void configure(
            Settings http,
            Settings dataset,
            Settings storage,
            Settings documentParse,
            Settings responseAggregation) {
        SHARED = create(http);
        DATASET = create(dataset);
        STORAGE = create(storage);
        DOCUMENT_PARSE = create(documentParse);
        RESPONSE_AGGREGATION =
                create(responseAggregation);
    }

    /**
     * 根据不可变配置创建隔离器。
     *
     * @param settings 隔离配置
     * @return 新隔离器
     */
    private static IoBulkhead create(Settings settings) {
        Settings value = java.util.Objects.requireNonNull(
                settings, "I/O bulkhead settings required");
        return new IoBulkhead(
                value.globalConcurrency(),
                value.perTargetConcurrency(),
                value.acquireTimeout(),
                value.maxTrackedTargets());
    }

    /**
     * I/O 隔离器的不可变配置。
     *
     * @param globalConcurrency 当前进程允许的 I/O 总并发
     * @param perTargetConcurrency 单个目标允许的 I/O 并发
     * @param acquireTimeout 等待许可的最长时间
     * @param maxTrackedTargets 最多单独跟踪的目标数量
     */
    public record Settings(
            int globalConcurrency,
            int perTargetConcurrency,
            Duration acquireTimeout,
            int maxTrackedTargets) {

        /**
         * 校验配置，避免应用启动后才暴露无效容量。
         *
         * @throws IllegalArgumentException 配置值无效时抛出
         */
        public Settings {
            if (globalConcurrency <= 0
                    || perTargetConcurrency <= 0
                    || acquireTimeout == null
                    || acquireTimeout.isNegative()
                    || acquireTimeout.isZero()
                    || maxTrackedTargets <= 0) {
                throw new IllegalArgumentException(
                        "I/O bulkhead settings must be positive");
            }
        }
    }

    /**
     * 将 URL 转换为稳定的协议和主机目标键。
     *
     * @param url 请求 URL
     * @return 目标键；URL 不合法时返回通用 HTTP 目标
     */
    public static String targetForUrl(String url) {
        if (!StringUtil.hasText(url)) {
            return "http:unknown";
        }
        try {
            URI uri = URI.create(url.trim());
            String host = uri.getHost();
            if (!StringUtil.hasText(host)) {
                return "http:unknown";
            }
            int port = uri.getPort();
            return "http:"
                    + host.toLowerCase(Locale.ROOT)
                    + (port < 0 ? "" : ":" + port);
        } catch (IllegalArgumentException exception) {
            return "http:unknown";
        }
    }

    /**
     * 在给定目标上申请一次阻塞 I/O 许可。
     *
     * @param target 目标标识
     * @return 使用完成后必须关闭的许可
     * @throws RetryableTriggerException 等待超时或线程被中断时抛出
     */
    public Permit acquire(String target) {
        String normalizedTarget = normalizeTarget(target);
        Semaphore targetSemaphore = targetSemaphore(normalizedTarget);
        long startedAt = System.nanoTime();
        boolean targetAcquired = false;
        boolean globalAcquired = false;
        try {
            targetAcquired = targetSemaphore.tryAcquire(acquireTimeoutNanos, TimeUnit.NANOSECONDS);
            if (!targetAcquired) {
                throw rejection(normalizedTarget, startedAt, null);
            }
            long remainingNanos = Math.max(
                    0L,
                    acquireTimeoutNanos - (System.nanoTime() - startedAt));
            globalAcquired = globalSemaphore.tryAcquire(remainingNanos, TimeUnit.NANOSECONDS);
            if (!globalAcquired) {
                throw rejection(normalizedTarget, startedAt, null);
            }
            totalWaitNanos.add(System.nanoTime() - startedAt);
            acquiredCount.increment();
            inFlightCount.increment();
            return new Permit(this, targetSemaphore);
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw rejection(normalizedTarget, startedAt, exception);
        } finally {
            if (!globalAcquired && targetAcquired) {
                targetSemaphore.release();
            }
        }
    }

    /**
     * 获取当前隔离器的轻量运行指标。
     *
     * @return 指标快照
     */
    public Snapshot snapshot() {
        return new Snapshot(
                acquiredCount.sum(),
                rejectedCount.sum(),
                inFlightCount.sum(),
                totalWaitNanos.sum(),
                globalSemaphore.availablePermits(),
                targetSemaphores.size());
    }

    /**
     * 释放成功申请的全局和目标许可。
     *
     * @param targetSemaphore 已申请的目标信号量
     */
    private void release(Semaphore targetSemaphore) {
        inFlightCount.decrement();
        globalSemaphore.release();
        targetSemaphore.release();
    }

    /**
     * 获取目标对应的信号量，并限制目标注册表体积。
     *
     * @param target 规范化目标
     * @return 目标信号量
     */
    private Semaphore targetSemaphore(String target) {
        Semaphore existing = targetSemaphores.get(target);
        if (existing != null) {
            return existing;
        }
        synchronized (targetRegistryLock) {
            existing = targetSemaphores.get(target);
            if (existing != null) {
                return existing;
            }
            if (targetSemaphores.size() >= maxTrackedTargets) {
                return overflowTargetSemaphore;
            }
            Semaphore created = new Semaphore(perTargetConcurrency, true);
            targetSemaphores.put(target, created);
            return created;
        }
    }

    /**
     * 构造可重新投递的限流异常并记录指标。
     *
     * @param target 目标标识
     * @param startedAt 等待开始时间
     * @param cause 原始异常
     * @return 可重新投递异常
     */
    private RetryableTriggerException rejection(String target, long startedAt, Throwable cause) {
        long waitedNanos = System.nanoTime() - startedAt;
        totalWaitNanos.add(waitedNanos);
        rejectedCount.increment();
        TimeoutException timeoutException = new TimeoutException(
                "I/O bulkhead is saturated for target " + target);
        if (cause != null) {
            timeoutException.initCause(cause);
        }
        logRejectionIfDue(target, waitedNanos);
        return new RetryableTriggerException("I/O 资源繁忙，请稍后重试", timeoutException);
    }

    /**
     * 对饱和日志限频，避免过载期间放大日志 I/O。
     *
     * @param target 目标标识
     * @param waitedNanos 本次等待纳秒数
     */
    private void logRejectionIfDue(String target, long waitedNanos) {
        long now = System.nanoTime();
        long previous = lastRejectionLogNanos.get();
        if ((previous != 0L && now - previous < REJECTION_LOG_INTERVAL_NANOS)
                || !lastRejectionLogNanos.compareAndSet(previous, now)) {
            return;
        }
        log.warn(
                "工作流 I/O 隔离器拒绝请求，target={}, waitedMs={}, inFlight={}, rejected={}",
                target,
                TimeUnit.NANOSECONDS.toMillis(waitedNanos),
                inFlightCount.sum(),
                rejectedCount.sum());
    }

    /**
     * 规范化目标键。
     *
     * @param target 原始目标
     * @return 非空目标键
     */
    private static String normalizeTarget(String target) {
        return StringUtil.hasText(target) ? target.trim() : "io:unknown";
    }

    /**
     * 读取正整数系统属性。
     *
     * @param name 属性名
     * @param defaultValue 默认值
     * @return 有效属性值或默认值
     */
    private static int positiveIntProperty(String name, int defaultValue) {
        try {
            int value = Integer.parseInt(System.getProperty(name, String.valueOf(defaultValue)));
            return value > 0 ? value : defaultValue;
        } catch (NumberFormatException exception) {
            return defaultValue;
        }
    }

    /**
     * 读取正长整数系统属性。
     *
     * @param name 属性名
     * @param defaultValue 默认值
     * @return 有效属性值或默认值
     */
    private static long positiveLongProperty(String name, long defaultValue) {
        try {
            long value = Long.parseLong(System.getProperty(name, String.valueOf(defaultValue)));
            return value > 0L ? value : defaultValue;
        } catch (NumberFormatException exception) {
            return defaultValue;
        }
    }

    /**
     * 按资源类别创建具有独立容量的隔离器。
     *
     * @param prefix 系统属性前缀
     * @param globalConcurrency 默认总并发
     * @param targetConcurrency 默认单目标并发
     * @param timeoutMillis 默认等待毫秒数
     * @param trackedTargets 默认目标上限
     * @return 独立隔离器
     */
    private static IoBulkhead lane(
            String prefix,
            int globalConcurrency,
            int targetConcurrency,
            long timeoutMillis,
            int trackedTargets) {
        return new IoBulkhead(
                positiveIntProperty(
                        prefix + ".max-concurrency",
                        globalConcurrency),
                positiveIntProperty(
                        prefix + ".per-target-max-concurrency",
                        targetConcurrency),
                Duration.ofMillis(positiveLongProperty(
                        prefix + ".acquire-timeout-ms",
                        timeoutMillis)),
                positiveIntProperty(
                        prefix + ".max-tracked-targets",
                        trackedTargets));
    }

    /**
     * 一次 I/O 许可，关闭时幂等释放资源。
     */
    public static final class Permit implements AutoCloseable {

        private final IoBulkhead owner;
        private final Semaphore targetSemaphore;
        private final AtomicBoolean closed = new AtomicBoolean();

        /**
         * 创建许可。
         *
         * @param owner 所属隔离器
         * @param targetSemaphore 目标信号量
         */
        private Permit(IoBulkhead owner, Semaphore targetSemaphore) {
            this.owner = owner;
            this.targetSemaphore = targetSemaphore;
        }

        /**
         * 幂等释放许可。
         */
        @Override
        public void close() {
            if (closed.compareAndSet(false, true)) {
                owner.release(targetSemaphore);
            }
        }
    }

    /**
     * I/O 隔离器的不可变运行指标快照。
     *
     * @param acquiredCount 已获取许可次数
     * @param rejectedCount 被拒绝次数
     * @param inFlightCount 当前执行中数量
     * @param totalWaitNanos 累计等待纳秒数
     * @param availableGlobalPermits 当前可用全局许可数
     * @param trackedTargetCount 当前独立跟踪目标数
     */
    public record Snapshot(
            long acquiredCount,
            long rejectedCount,
            long inFlightCount,
            long totalWaitNanos,
            int availableGlobalPermits,
            int trackedTargetCount) {
    }
}
