from core.models.system.SystemState import SystemState from core.models.session.SessionState import SessionState from dataclasses import dataclass, field from typing import Optional @dataclass class GlobalState: systemwide_active: bool = False firewalled: bool = False dns_set: bool = False systemwide_profile_id: Optional[int] = None session_profile_ids: list[int] = field(default_factory=list) verified: bool = False class ProfileStateRepository: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(ProfileStateRepository, cls).__new__(cls) cls._instance._initialize() return cls._instance def _initialize(self): self.global_state = GlobalState() # Initial load: Sync memory with whatever is currently on disk sys = SystemState.get() if sys: self.global_state.systemwide_profile_id = sys.profile_id self.global_state.firewalled = sys.firewalled self.global_state.dns_set = sys.dns_set # Load existing session IDs from disk into the list self.global_state.session_profile_ids = [s.id for s in SessionState.all()] def sync_system_state(self, state: SystemState): """Called by SystemStateController to update memory cache.""" self.global_state.systemwide_profile_id = state.profile_id self.global_state.firewalled = state.firewalled self.global_state.dns_set = state.dns_set def sync_session_added(self, profile_id: int): """Called by SessionStateController to update memory cache.""" if profile_id not in self.global_state.session_profile_ids: self.global_state.session_profile_ids.append(profile_id) def sync_session_removed(self, profile_id: int): """Called by SessionStateController to update memory cache.""" if profile_id in self.global_state.session_profile_ids: self.global_state.session_profile_ids.remove(profile_id) def is_profile_active(self, profile_id: int) -> bool: # Check system profile if self.global_state.systemwide_active and self.global_state.systemwide_profile_id == profile_id: return True # Check session profiles return profile_id in self.global_state.session_profile_ids