# edited by glg import threading import time from pypos.core.base_service import BaseService from pypos.modules.sinkronisasi.services.network_probe_adapter_service import ( NetworkProbeAdapterService, ) from pypos.modules.sinkronisasi.config.sync_config import ( get_network_orchestrator_policy, ) class NetworkOrchestratorService(BaseService): STATE_ONLINE_STABLE = "ONLINE_STABLE" STATE_ONLINE_DEGRADED = "ONLINE_DEGRADED" STATE_INTERNET_ONLY = "INTERNET_ONLY" STATE_OFFLINE = "OFFLINE" PROCESS_SYNC = "sync" PROCESS_EXPORT = "export" PROCESS_SETTLEMENT_DIRECT = "settlement_direct" _STATE_LOCK = threading.Lock() _CACHE_UNTIL_TS = 0.0 _LAST_EVALUATION = {} _FAIL_STREAK = 0 _SUCCESS_STREAK = 0 _LAST_GOOD_TS = 0.0 def __init__( self, probe_service=None, policy_getter=get_network_orchestrator_policy, now_fn=None, ): super().__init__() self.probe_service = probe_service or NetworkProbeAdapterService() self.policy_getter = policy_getter self.now_fn = now_fn or time.monotonic def _now(self) -> float: try: return float(self.now_fn()) except (TypeError, ValueError, RuntimeError): return float(time.monotonic()) @staticmethod def _to_bool(value, default=False) -> bool: if isinstance(value, bool): return bool(value) text = str(value or "").strip().lower() if text in {"1", "true", "yes", "on"}: return True if text in {"0", "false", "no", "off"}: return False return bool(default) @staticmethod def _to_int(value, default=0, minimum=0) -> int: try: parsed = int(value) except (TypeError, ValueError): parsed = int(default) return max(int(minimum), parsed) @staticmethod def _to_float(value, default=0.0, minimum=0.0) -> float: try: parsed = float(value) except (TypeError, ValueError): parsed = float(default) return max(float(minimum), parsed) def _read_policy(self): raw = self.policy_getter() if callable(self.policy_getter) else {} cfg = raw if isinstance(raw, dict) else {} process_allow_internet_only = cfg.get("process_allow_internet_only") if not isinstance(process_allow_internet_only, dict): process_allow_internet_only = {} process_map = { self.PROCESS_SYNC: self._to_bool(process_allow_internet_only.get(self.PROCESS_SYNC), default=False), self.PROCESS_EXPORT: self._to_bool(process_allow_internet_only.get(self.PROCESS_EXPORT), default=False), self.PROCESS_SETTLEMENT_DIRECT: self._to_bool( process_allow_internet_only.get(self.PROCESS_SETTLEMENT_DIRECT), default=False, ), } policy = { "enabled": self._to_bool(cfg.get("enabled"), default=True), "probe_cache_ttl_sec": self._to_float(cfg.get("probe_cache_ttl_sec"), default=2.0, minimum=0.0), "fail_threshold": self._to_int(cfg.get("fail_threshold"), default=3, minimum=1), "recover_threshold": self._to_int(cfg.get("recover_threshold"), default=2, minimum=1), "grace_seconds": self._to_float(cfg.get("grace_seconds"), default=45.0, minimum=0.0), "allow_unstable_operations": self._to_bool(cfg.get("allow_unstable_operations"), default=True), "allow_degraded_without_server": self._to_bool( cfg.get("allow_degraded_without_server"), default=False, ), "process_allow_internet_only": process_map, "server_probe_timeout_sec": self._to_float(cfg.get("server_probe_timeout_sec"), default=1.0, minimum=0.2), "internet_probe_timeout_sec": self._to_float( cfg.get("internet_probe_timeout_sec"), default=0.8, minimum=0.2, ), "internet_probe_urls": cfg.get("internet_probe_urls") if isinstance(cfg.get("internet_probe_urls"), list) else [], "health_check_enabled": self._to_bool(cfg.get("health_check_enabled"), default=False), "health_endpoint": str(cfg.get("health_endpoint") or "").strip(), "health_timeout_sec": self._to_float(cfg.get("health_timeout_sec"), default=1.0, minimum=0.2), } return policy def _build_probe_policy(self, policy): return { "server_probe_timeout_sec": float(policy.get("server_probe_timeout_sec") or 1.0), "internet_probe_timeout_sec": float(policy.get("internet_probe_timeout_sec") or 0.8), "internet_probe_urls": list(policy.get("internet_probe_urls") or []), "health_check_enabled": bool(policy.get("health_check_enabled")), "health_endpoint": str(policy.get("health_endpoint") or "").strip(), "health_timeout_sec": float(policy.get("health_timeout_sec") or 1.0), } def _probe_snapshot(self, policy): probe_policy = self._build_probe_policy(policy) snapshot = self.probe_service.probe_network_snapshot(policy=probe_policy) if isinstance(snapshot, dict): return snapshot return {} def _classify_state(self, raw_state, server_online, policy, now_ts): fail_threshold = int(policy.get("fail_threshold") or 3) recover_threshold = int(policy.get("recover_threshold") or 2) grace_seconds = float(policy.get("grace_seconds") or 0.0) allow_unstable = bool(policy.get("allow_unstable_operations")) with self._STATE_LOCK: if str(raw_state) == NetworkProbeAdapterService.RAW_SERVER_OK: self._SUCCESS_STREAK += 1 self._FAIL_STREAK = 0 self._LAST_GOOD_TS = now_ts else: self._SUCCESS_STREAK = 0 self._FAIL_STREAK += 1 fail_streak = int(self._FAIL_STREAK) success_streak = int(self._SUCCESS_STREAK) last_good_ts = float(self._LAST_GOOD_TS or 0.0) within_grace = False if grace_seconds > 0 and last_good_ts > 0: within_grace = (now_ts - last_good_ts) <= grace_seconds if str(raw_state) == NetworkProbeAdapterService.RAW_SERVER_OK: state = self.STATE_ONLINE_STABLE if success_streak >= recover_threshold else self.STATE_ONLINE_DEGRADED elif str(raw_state) == NetworkProbeAdapterService.RAW_INTERNET_ONLY: if allow_unstable and within_grace and fail_streak < fail_threshold: state = self.STATE_ONLINE_DEGRADED else: state = self.STATE_INTERNET_ONLY else: if allow_unstable and within_grace and fail_streak < fail_threshold: state = self.STATE_ONLINE_DEGRADED else: state = self.STATE_OFFLINE return state, fail_streak, success_streak, within_grace def evaluate(self, force_probe=False): policy = self._read_policy() if not bool(policy.get("enabled")): return { "enabled": False, "state": self.STATE_ONLINE_STABLE, "raw_state": "DISABLED", "server_online": True, "internet_online": True, "fail_streak": 0, "success_streak": 0, "within_grace": True, "snapshot": {}, "cached": False, } now_ts = self._now() with self._STATE_LOCK: cache_until = float(self._CACHE_UNTIL_TS or 0.0) cached_payload = dict(self._LAST_EVALUATION or {}) if not bool(force_probe) and cached_payload and now_ts <= cache_until: cached_payload["cached"] = True return cached_payload snapshot = self._probe_snapshot(policy) raw_state = str(snapshot.get("raw_state") or "") server_online = bool(snapshot.get("server_online")) internet_online = bool(snapshot.get("internet_online")) state, fail_streak, success_streak, within_grace = self._classify_state( raw_state=raw_state, server_online=server_online, policy=policy, now_ts=now_ts, ) payload = { "enabled": True, "state": state, "raw_state": raw_state, "server_online": bool(server_online), "internet_online": bool(internet_online), "fail_streak": int(fail_streak), "success_streak": int(success_streak), "within_grace": bool(within_grace), "snapshot": snapshot if isinstance(snapshot, dict) else {}, "cached": False, } ttl = float(policy.get("probe_cache_ttl_sec") or 0.0) with self._STATE_LOCK: self._LAST_EVALUATION = dict(payload) self._CACHE_UNTIL_TS = now_ts + max(0.0, ttl) return payload def can_run(self, process, force_probe=False): process_name = str(process or "").strip().lower() or self.PROCESS_SYNC policy = self._read_policy() snapshot = self.evaluate(force_probe=force_probe) state = str(snapshot.get("state") or self.STATE_OFFLINE) server_online = bool(snapshot.get("server_online")) allow = False reason = "" if state == self.STATE_ONLINE_STABLE: allow = True reason = "online_stable" elif state == self.STATE_ONLINE_DEGRADED: if server_online: allow = bool(policy.get("allow_unstable_operations")) reason = "online_degraded_server_reachable" else: allow = bool(policy.get("allow_unstable_operations")) and bool( policy.get("allow_degraded_without_server") ) reason = "online_degraded_server_unreachable" elif state == self.STATE_INTERNET_ONLY: allow_map = policy.get("process_allow_internet_only") or {} allow = bool(allow_map.get(process_name, False)) reason = "internet_only" else: allow = False reason = "offline" result = dict(snapshot) result.update( { "process": process_name, "allow": bool(allow), "reason": reason, } ) return result