"""跨进程熔断门: CHS provider_gate 五操作 Lua 逐字移植(D3 双后端;M2 设计 §3)。

语义蓝本 `reference/CHSAnalyzer/app/coordination/{scripts,provider_gate}.py`:
每源一个 HASH(state/epoch/failures/open_until/probe_until/probe_owner),
每操作单条 Lua 原子。保真点: epoch **仅在 record_failure 开断时 +1**
(scripts.py:180);探针/force_open failures 顶格(:173-178);fencing 判据
两分支(:119-123/:157-161);release_probe 置 open_until=now 让下家立即
接管(:210-216);探针租约 = probe_until 绝对 ms + 惰性重发,无看门狗;
时钟回拨经 math.max clamp。已声明偏离: key 前缀 `provider_gate:` →
`pgw:gate:`;`LimiterError` → `GovernanceBackendError`。
"""

from __future__ import annotations

import math
from typing import TYPE_CHECKING

from redis.exceptions import RedisError

from polygateway.errors import GovernanceBackendError
from polygateway.ports import GateDecision, GateState, GateUpdate

if TYPE_CHECKING:
    from redis.asyncio import Redis

    from polygateway.types import BreakerConfig

# KEYS: provider_hash ; ARGV: owner probe_ttl_ms(CHS scripts.py:82-106)
# 返回: allowed state epoch is_probe probe_owner retry_after_ms
TRY_ENTER = """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
local open_until = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
local probe_until = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')

if state == 'closed' then
    return {1, state, epoch, 0, '', 0}
end
if state == 'open' and now < open_until then
    return {0, state, epoch, 0, '', open_until - now}
end
if state == 'half_open' and now < probe_until then
    -- 探针在途: 无确定的可再试时刻 → 0(issue #14,与 memory `_remaining` 同口径)
    return {0, state, epoch, 0, '', 0}
end

local next_probe_until = now + tonumber(ARGV[2])
redis.call('HSET', KEYS[1],
    'state', 'half_open',
    'probe_owner', ARGV[1],
    'probe_until', next_probe_until)
return {1, 'half_open', epoch, 1, ARGV[1], 0}
"""

# M2.5 窗口/退避公共片段(拼接进 success/failure 脚本;Lua 脚本间无法共享函数)
_WINDOW_HELPERS = """
local function rotate_window(key, now, half_ms)
    local wid = math.floor(now / half_ms)
    local cur = tonumber(redis.call('HGET', key, 'win_id') or '-1')
    if wid ~= cur then
        if wid == cur + 1 then
            redis.call('HSET', key,
                'a1', redis.call('HGET', key, 'a0') or '0',
                'f1', redis.call('HGET', key, 'f0') or '0')
        else
            redis.call('HSET', key, 'a1', 0, 'f1', 0)
        end
        redis.call('HSET', key, 'a0', 0, 'f0', 0, 'win_id', wid)
    end
end
local function cooldown_eff(key, cooldown_ms, max_cooldown_ms)
    local streak = math.max(1, math.min(
        tonumber(redis.call('HGET', key, 'reopen_streak') or '0'), 16))
    return math.floor(math.min(cooldown_ms * 2 ^ (streak - 1), max_cooldown_ms))
end
"""

# KEYS: provider_hash(CHS :110-143 + M2.5 窗口/衰减)
# ARGV: expected_epoch is_probe probe_owner count_attempt half_window_ms cooldown_ms max_cooldown_ms
# 返回: applied state epoch failures retry_after_ms
RECORD_SUCCESS = (
    _WINDOW_HELPERS
    + """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
local is_probe = tonumber(ARGV[2])
local matches = false
if is_probe == 1 then
    matches = state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[3]
else
    matches = state == 'closed' and epoch == tonumber(ARGV[1])
end

if matches then
    if tonumber(ARGV[4]) == 1 then
        rotate_window(KEYS[1], now, tonumber(ARGV[5]))
        redis.call('HINCRBY', KEYS[1], 'a0', 1)
    end
    -- streak 衰减: CLOSED 稳定满 2×cooldown_eff 后的首次成功归零(设计 §3.1)
    local streak = tonumber(redis.call('HGET', KEYS[1], 'reopen_streak') or '0')
    local since = tonumber(redis.call('HGET', KEYS[1], 'closed_since') or '0')
    if streak > 0 and since > 0
        and now - since >= 2 * cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[7])) then
        redis.call('HSET', KEYS[1], 'reopen_streak', 0)
    end
    if is_probe == 1 then
        redis.call('HSET', KEYS[1], 'closed_since', now)
    end
    redis.call('HSET', KEYS[1],
        'state', 'closed',
        'failures', 0,
        'epoch', epoch,
        'open_until', 0,
        'probe_owner', '',
        'probe_until', 0)
    return {1, 'closed', epoch, 0, 0}
end

local deadline = 0
if state == 'open' then
    deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
end
return {0, state, epoch, failures, math.max(deadline - now, 0)}
"""
)

# KEYS: provider_hash(CHS :148-199 + M2.5 双通道/退避)
# ARGV: expected_epoch is_probe probe_owner force_open threshold cooldown_ms
#       rate_limited min_calls fail_rate half_window_ms max_cooldown_ms
# 返回: applied state epoch failures retry_after_ms
RECORD_FAILURE = (
    _WINDOW_HELPERS
    + """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
local is_probe = tonumber(ARGV[2])
local matches = false
if is_probe == 1 then
    matches = state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[3]
else
    matches = state == 'closed' and epoch == tonumber(ARGV[1])
end

if not matches then
    local deadline = 0
    if state == 'open' then
        deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
    end
    return {0, state, epoch, failures, math.max(deadline - now, 0)}
end

local force_open = tonumber(ARGV[4])
if tonumber(ARGV[7]) == 1 and force_open == 0 then
    -- 429 = 背压不是故障(M2.5 设计 §3.1): 两通道均不计;
    -- 探针撞 429 按无果归还语义放下家,不挂租约
    if is_probe == 1 then
        redis.call('HSET', KEYS[1],
            'state', 'open',
            'open_until', now,
            'probe_owner', '',
            'probe_until', 0)
        return {1, 'open', epoch, failures, 0}
    end
    return {1, state, epoch, failures, 0}
end

local threshold = tonumber(ARGV[5])
local open_via_rate = false
local window_healthy = false
if force_open == 0 and is_probe == 0 then
    rotate_window(KEYS[1], now, tonumber(ARGV[10]))
    redis.call('HINCRBY', KEYS[1], 'a0', 1)
    redis.call('HINCRBY', KEYS[1], 'f0', 1)
    local attempts = tonumber(redis.call('HGET', KEYS[1], 'a0') or '0')
        + tonumber(redis.call('HGET', KEYS[1], 'a1') or '0')
    local fails_w = tonumber(redis.call('HGET', KEYS[1], 'f0') or '0')
        + tonumber(redis.call('HGET', KEYS[1], 'f1') or '0')
    if attempts >= tonumber(ARGV[8]) then
        if fails_w / attempts >= tonumber(ARGV[9]) then
            open_via_rate = true
        else
            -- 窗口证据充足且健康: 连败是噪声,抑制连续通道(迭代 6)
            window_healthy = true
        end
    end
end

if force_open == 1 or is_probe == 1 then
    failures = threshold
else
    failures = failures + 1
end
local open_via_streak = failures >= threshold and not window_healthy
if force_open == 1 or is_probe == 1 or open_via_rate or open_via_streak then
    -- 递增 streak 的只有率通道开路与探针失败重开(C1: 连续通道/force_open 不递增)
    if open_via_rate or is_probe == 1 then
        redis.call('HINCRBY', KEYS[1], 'reopen_streak', 1)
    end
    if open_via_rate then
        failures = math.max(failures, threshold)
    end
    epoch = epoch + 1
    local eff = cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[11]))
    redis.call('HSET', KEYS[1],
        'state', 'open',
        'failures', failures,
        'epoch', epoch,
        'open_until', now + eff,
        'probe_owner', '',
        'probe_until', 0)
    redis.call('HDEL', KEYS[1], 'closed_since')
    return {1, 'open', epoch, failures, eff}
end
redis.call('HSET', KEYS[1],
    'state', 'closed',
    'failures', failures,
    'epoch', epoch,
    'open_until', 0,
    'probe_owner', '',
    'probe_until', 0)
return {1, 'closed', epoch, failures, 0}
"""
)

# KEYS: provider_hash ; ARGV: expected_epoch probe_owner(CHS :203-225)
# 返回: applied state epoch failures retry_after_ms
RELEASE_PROBE = """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
if state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[2] then
    redis.call('HSET', KEYS[1],
        'state', 'open',
        'open_until', now,
        'probe_owner', '',
        'probe_until', 0)
    return {1, 'open', epoch, failures, 0}
end
local deadline = 0
if state == 'open' then
    deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
end
return {0, state, epoch, failures, math.max(deadline - now, 0)}
"""

# KEYS: 若干 provider hash;返回最早可尝试等待毫秒(CHS :228-245)
RETRY_AFTER = """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local minimum = nil
for _, key in ipairs(KEYS) do
    local state = redis.call('HGET', key, 'state') or 'closed'
    local remaining = 0
    if state == 'open' then
        local deadline = tonumber(redis.call('HGET', key, 'open_until') or '0')
        remaining = math.max(deadline - now, 0)
    end
    if minimum == nil or remaining < minimum then minimum = remaining end
end
return minimum or 0
"""


class RedisGate:
    """以 Redis HASH + Lua 原子维护每源健康状态;同 scope 多进程共享。"""

    def __init__(self, *, config: BreakerConfig, redis: Redis, scope: str) -> None:
        if not scope.strip():
            raise ValueError("scope 不能为空")
        self._scope = scope.lower()
        self._threshold = config.fail_threshold
        self._cooldown_ms = math.ceil(config.cooldown_s * 1000)
        self._probe_ttl_ms = math.ceil(config.probe_ttl_s * 1000)
        self._max_cooldown_ms = math.ceil(config.max_cooldown_s * 1000)
        self._half_window_ms = max(1, math.ceil(config.window_s * 500))  # 半窗
        self._min_calls = config.min_calls
        self._fail_rate = config.fail_rate
        self._redis = redis
        self._owns_client = False
        self._try_enter_lua = redis.register_script(TRY_ENTER)
        self._success_lua = redis.register_script(RECORD_SUCCESS)
        self._failure_lua = redis.register_script(RECORD_FAILURE)
        self._release_lua = redis.register_script(RELEASE_PROBE)
        self._retry_after_lua = redis.register_script(RETRY_AFTER)

    @classmethod
    def from_url(cls, url: str, **kwargs) -> RedisGate:
        """自建并持有 Redis 客户端(aclose 时代关);共享后端请直接注入 redis。"""
        import redis.asyncio as aioredis

        gate = cls(redis=aioredis.from_url(url), **kwargs)
        gate._owns_client = True
        return gate

    def _key(self, source_name: str) -> str:
        if not source_name.strip():
            raise ValueError("source_name 不能为空")
        return f"pgw:gate:{self._scope}:{source_name}"

    @staticmethod
    def _state(value: object) -> GateState:
        if isinstance(value, bytes):
            value = value.decode("utf-8")
        return GateState(str(value))

    @classmethod
    def _decision(cls, source_name: str, result: list[object]) -> GateDecision:
        """解析 try_enter 六元返回;GateDecision 自带不变式校验兜底移植正确性。"""
        is_probe = int(result[3]) == 1
        owner = result[4]
        if isinstance(owner, bytes):
            owner = owner.decode("utf-8")
        return GateDecision(
            source_name=source_name,
            allowed=int(result[0]) == 1,
            state=cls._state(result[1]),
            epoch=int(result[2]),
            is_probe=is_probe,
            probe_owner=str(owner) if is_probe else None,
            retry_after_s=int(result[5]) / 1000.0,
        )

    @classmethod
    def _update(cls, result: list[object]) -> GateUpdate:
        return GateUpdate(
            applied=int(result[0]) == 1,
            state=cls._state(result[1]),
            epoch=int(result[2]),
            failure_count=int(result[3]),
            retry_after_s=int(result[4]) / 1000.0,
        )

    @staticmethod
    def _entry_args(entry: GateDecision) -> list[object]:
        """提取 fencing 令牌;拒绝用被拒决定伪造写回(CHS provider_gate.py:94-103)。"""
        if not entry.allowed:
            raise ValueError("被拒决定不得写回")
        return [entry.epoch, 1 if entry.is_probe else 0, entry.probe_owner or ""]

    async def try_enter(self, source_name: str, owner: str) -> GateDecision:
        """健康普通准入;冷却到期/探针租约过期时原子授予唯一探针。"""
        if not owner.strip():
            raise ValueError("owner 不能为空")
        try:
            result = await self._try_enter_lua(
                keys=[self._key(source_name)], args=[owner, self._probe_ttl_ms]
            )
        except RedisError as exc:
            raise GovernanceBackendError(
                f"熔断后端 try_enter 失败: {exc}", scope=self._scope
            ) from exc
        return self._decision(source_name, result)

    async def record_success(
        self, entry: GateDecision, *, count_attempt: bool = True
    ) -> GateUpdate:
        args = self._entry_args(entry)
        args.extend(
            [
                1 if count_attempt else 0,
                self._half_window_ms,
                self._cooldown_ms,
                self._max_cooldown_ms,
            ]
        )
        try:
            result = await self._success_lua(keys=[self._key(entry.source_name)], args=args)
        except RedisError as exc:
            raise GovernanceBackendError(
                f"熔断后端 record_success 失败: {exc}", scope=self._scope
            ) from exc
        return self._update(result)

    async def record_failure(
        self, entry: GateDecision, reason: str, force_open: bool
    ) -> GateUpdate:
        args = self._entry_args(entry)
        args.extend(
            [
                1 if force_open else 0,
                self._threshold,
                self._cooldown_ms,
                1 if reason == "rate_limited" else 0,
                self._min_calls,
                self._fail_rate,
                self._half_window_ms,
                self._max_cooldown_ms,
            ]
        )
        try:
            result = await self._failure_lua(keys=[self._key(entry.source_name)], args=args)
        except RedisError as exc:
            raise GovernanceBackendError(
                f"熔断后端 record_failure 失败: {exc}", scope=self._scope
            ) from exc
        return self._update(result)

    async def release_probe(self, entry: GateDecision) -> GateUpdate:
        """探针无果归还: 源保持 OPEN 且 open_until=now,下家立即可接管;幂等。"""
        if not (entry.allowed and entry.is_probe and entry.probe_owner is not None):
            raise ValueError("release_probe 只接受在途探针决定")
        try:
            result = await self._release_lua(
                keys=[self._key(entry.source_name)], args=[entry.epoch, entry.probe_owner]
            )
        except RedisError as exc:
            raise GovernanceBackendError(
                f"熔断后端 release_probe 失败: {exc}", scope=self._scope
            ) from exc
        return self._update(result)

    async def retry_after_s(self, sources: tuple[str, ...]) -> float:
        """集合中最早可尝试等待秒数;健康/到期返回 0;空集合 ValueError(契约)。"""
        if not sources:
            raise ValueError("sources 不能为空")
        try:
            result = await self._retry_after_lua(keys=[self._key(s) for s in sources])
        except RedisError as exc:
            raise GovernanceBackendError(
                f"熔断后端 retry_after_s 失败: {exc}", scope=self._scope
            ) from exc
        return int(result) / 1000.0

    async def aclose(self) -> None:
        """幂等释放自建客户端;注入的客户端归注入方管理。"""
        if self._owns_client:
            self._owns_client = False
            await self._redis.aclose()
