# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
# OpenOLC is licensed under Mulan PSL v2.
# You can use this software according to the terms and conditions of the Mulan PSL v2.
# You may obtain a copy of Mulan PSL v2 at:
#         `http://license.coscl.org.cn/MulanPSL2`
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
# See the Mulan PSL v2 for more details.

import threading
from typing import Dict, Optional, List

import logging

from olc.config.properties_loader import OlcConfigManager

from olc.bean.olc_config_rule import FlowPolicy, AdmissionPolicy, BasePolicy
from olc.bean.tag_group import TagGroup, SubGroup
from olc.cache.safe__ttl_cache import SafeTTLCache
from olc.utils.str_utils import StrUtils

logger = logging.getLogger(__name__)


class OlcRuleCache:
    def __init__(self, domain: str = ""):
        self.domain = domain
        self.lock = threading.RLock()
        # 使用普通字典配合锁来实现线程安全的映射
        self.tag_groups: Dict[str, TagGroup] = {}
        self.bind_flow_relation: Dict[str, FlowPolicy] = {}
        self.bind_admission_relation: Dict[str, AdmissionPolicy] = {}
        max_size = OlcConfigManager.get_instance().get_sdk_config().sub_group_cache_size
        self.sub_groups: SafeTTLCache[SubGroup] = SafeTTLCache(maxsize=max_size, ttl=3600)

    def __repr__(self):
        return (f"OlcRuleCache(domain={self.domain}, "
                f"tag_groups={dict(self.tag_groups)}, "
                f"flow_policies={dict(self.bind_flow_relation)}, "
                f"admission_policies={dict(self.bind_admission_relation)})")

    def __str__(self):
        return self.__repr__()

    def update_all(self, new_olc_rule_cache: Optional["OlcRuleCache"]):
        if new_olc_rule_cache is None:
            return
        try:
            with self.lock:
                self.tag_groups = dict(new_olc_rule_cache.tag_groups)
                self.bind_flow_relation = dict(new_olc_rule_cache.bind_flow_relation)
                self.bind_admission_relation = dict(
                    new_olc_rule_cache.bind_admission_relation
                )
        except Exception as e:
            logging.warning(f"Olc rule cache update failed: {e}")

    def add_tag_group(self, tag_groups: Optional[list[TagGroup]]):
        if tag_groups is None:
            return

        with self.lock:
            for tag_group in tag_groups:
                self.tag_groups[tag_group.name] = tag_group

    def add_policy(
        self, tag_groups: Optional[list[TagGroup]], policy: Optional[BasePolicy]
    ):
        if tag_groups is None or policy is None:
            return

        with self.lock:
            for tag_group in tag_groups:
                if isinstance(policy, FlowPolicy):
                    self.bind_flow_relation[tag_group.name] = policy
                if isinstance(policy, AdmissionPolicy):
                    self.bind_admission_relation[tag_group.name] = policy

    def add_flow_policy(
        self, tag_groups: Optional[list[TagGroup]], policy: Optional[FlowPolicy]
    ):
        """
        添加流策略
        """
        self.add_policy(tag_groups, policy)

    def add_admission_policy(
        self, tag_groups: Optional[list[TagGroup]], policy: Optional[AdmissionPolicy]
    ):
        """
        添加准入策略
        """
        self.add_policy(tag_groups, policy)

    def query_tag_group(self, tag_group_name: str) -> Optional[TagGroup]:
        if tag_group_name is None:
            return None
        with self.lock:
            group = self.tag_groups.get(tag_group_name)
            if group is None:
                group = self.sub_groups.get(tag_group_name)
            return group

    def add_sub_group(self, sub_group: SubGroup):
        if sub_group is None:
            return
        with self.lock:
            self.sub_groups.put(sub_group.name, sub_group)


class OlcRuleManager:
    """
    规则管理器
    """

    _instance = None
    _lock = threading.RLock()

    def __init__(self, domain: str = ""):
        self.rule_cache: Optional[OlcRuleCache] = OlcRuleCache(domain=domain)

    @classmethod
    def get_instance(cls) -> "OlcRuleManager":
        """
        获取单例实例
        """
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    def get_rule_cache(self) -> Optional[OlcRuleCache]:
        return self.rule_cache

    def config_update(self, olc_rule_cache: Optional[OlcRuleCache]):
        self.rule_cache.update_all(olc_rule_cache)

    def get_group_exclude_sub_group(self) -> Optional[list[TagGroup]]:
        if self.rule_cache is not None:
            return list(self.rule_cache.tag_groups.values())
        return None

    def get_flow_policy_by_group_name(self, name: str) -> Optional[FlowPolicy]:
        if StrUtils.is_empty(name) or self.rule_cache is None:
            return None
        return self.rule_cache.bind_flow_relation.get(name, None)

    def get_admission_policy_by_group_name(self, name: str) -> Optional[AdmissionPolicy]:
        if StrUtils.is_empty(name) or self.rule_cache is None:
            return None
        return self.rule_cache.bind_admission_relation.get(name, None)

    def get_group_from_cache(self, name: str) -> Optional[TagGroup]:
        if StrUtils.is_empty(name) or self.rule_cache is None:
            return None
        return self.rule_cache.query_tag_group(name)

    def is_group_policy_dup(self, sub_group: SubGroup) -> bool:
        if self.rule_cache is None:
            return False
        cache = self.get_group_from_cache(sub_group.name)
        return cache is not None

    def add_sub_group(self, group: SubGroup):
        if self.rule_cache is not None:
            self.rule_cache.add_sub_group(group)

    def refresh_sub_group(self, groups: List[TagGroup]):
        """
        刷新子组过期时间，防止过期
        """
        if self.rule_cache is not None:
            for group in groups:
                self.rule_cache.sub_groups.refresh(group.name)