Fix (thread safety) inMemory.

'''

Class: InMemoryAppStorage

'''

from threading import Lock

from typing import Dict, List, Optional

from sunrise6g_opensdk.edgecloud.adapters.aeros.storageManagement.appStorageManager \

    import AppStorageManager

from threading import RLock 

from typing import Dict, List, Optional, Union

from sunrise6g_opensdk.edgecloud.adapters.aeros.storageManagement.appStorageManager import AppStorageManager

from sunrise6g_opensdk.edgecloud.core.camara_schemas import AppInstanceInfo

from sunrise6g_opensdk.edgecloud.adapters.aeros import config

from sunrise6g_opensdk.logger import setup_logger

# import copy  # optional if you want deep copies

class SingletonMeta(type):

    """Thread-safe Singleton metaclass."""

    _instances: Dict[type, object] = {}

    _lock = Lock()

    _lock = RLock()

    def __call__(cls, *args, **kwargs):

        # Double-checked locking pattern

        if cls not in cls._instances:

            with cls._lock:

                if cls not in cls._instances:

class InMemoryAppStorage(AppStorageManager, metaclass=SingletonMeta):

    '''

    In-memory implementation of the AppStorageManager interface.

    '''

    """

    In-memory implementation of the AppStorageManager interface (process-wide singleton).

    """

    def __init__(self):

        # Make __init__ idempotent so repeated calls don't reset state

        if getattr(self, "_initialized", False):

            return

        if config.DEBUG:

        # Initialize logger always; emit debug message conditionally

        self.logger = setup_logger()

        if config.DEBUG:

            self.logger.info("Using InMemoryStorage")

        self._lock = RLock()

        self._apps: Dict[str, Dict] = {}

        self._deployed: Dict[str, List[AppInstanceInfo]] = {}

        self._stopped: Dict[str, List[str]] = {}

        self._initialized = True

    def reset(self) -> None:

        '''

        Helpful for unit tests to clear global state

        '''

        with self._lock:

            self._apps.clear()

            self._deployed.clear()

            self._stopped.clear()

    # --- Apps ---

    def store_app(self, app_id: str, manifest: Dict) -> None:

        with self._lock:

            self._apps[app_id] = manifest

    def get_app(self, app_id: str) -> Optional[Dict]:

        with self._lock:

            return self._apps.get(app_id)

    def app_exists(self, app_id: str) -> bool:

        with self._lock:

            return app_id in self._apps

    def list_apps(self) -> List[Dict]:

        return list(self._apps.values())

        with self._lock:

            # If you want full isolation, use deepcopy:

            # return [copy.deepcopy(m) for m in self._apps.values()]

            return [dict(m) for m in self._apps.values()]

    def delete_app(self, app_id: str) -> None:

        with self._lock:

            self._apps.pop(app_id, None)

    # --- Deployments ---

    def store_deployment(self, app_instance: AppInstanceInfo) -> None:

        app_id = str(app_instance.appId)

        if app_id not in self._deployed:

            self._deployed[app_id] = []

        self._deployed[app_id].append(app_instance)

        with self._lock:

            aid = str(app_instance.appId)

            self._deployed.setdefault(aid, []).append(app_instance)

    # Conform to interface -> Dict[str, List[str]]

    def get_deployments(self,

                        app_id: Optional[str] = None) -> List[AppInstanceInfo]:

                        app_id: Optional[str] = None) -> Dict[str, List[str]]:

        with self._lock:

            if app_id:

            return self._deployed.get(app_id, [])

        all_instances = []

        for instances in self._deployed.values():

            all_instances.extend(instances)

        return all_instances

    def find_deployments(self, app_id=None, app_instance_id=None, region=None):

        result = []

        for instances in self._deployed.values():  # iterate lists of instances

            for instance in instances:  # iterate individual AppInstanceInfo objects

                if app_id and str(instance.appId) != app_id:

                ids = [

                    str(i.appInstanceId)

                    for i in self._deployed.get(app_id, [])

                ]

                return {app_id: ids}

            return {

                aid: [str(i.appInstanceId) for i in insts]

                for aid, insts in self._deployed.items()

            }

    def find_deployments(

        self,

        app_id: Optional[str] = None,

        app_instance_id: Optional[str] = None,

        region: Optional[str] = None,

    ) -> List[AppInstanceInfo]:

        with self._lock:

            # Fast path by instance id

            if app_instance_id:

                for insts in self._deployed.values():

                    for inst in insts:

                        if str(inst.appInstanceId) == app_instance_id:

                            if app_id and str(inst.appId) != app_id:

                                return []

                            if region is not None and getattr(

                                    inst, "region", None) != region:

                                return []

                            return [inst]

                return []

            results: List[AppInstanceInfo] = []

            for aid, insts in self._deployed.items():

                if app_id and aid != app_id:

                    continue

                if app_instance_id and str(

                        instance.appInstanceId) != app_instance_id:

                for inst in insts:

                    if region is not None and getattr(inst, "region",

                                                      None) != region:

                        continue

                # Region filtering can go here if needed

                result.append(instance)

        return result

                    results.append(inst)

            return results

    def remove_deployment(self, app_instance_id: str) -> Optional[str]:

        for app_id, instances in self._deployed.items():

            for instance in instances:

                if str(instance.appInstanceId) == app_instance_id:

                    instances.remove(instance)

                    if not instances:

                        del self._deployed[app_id]

                    return app_id  # return the app_id that had this instance

        with self._lock:

            for aid, insts in list(

                    self._deployed.items()):  # iterate over a copy of items

                for idx, inst in enumerate(insts):

                    if str(inst.appInstanceId) == app_instance_id:

                        insts.pop(idx)

                        if not insts:

                            self._deployed.pop(aid, None)

                        return aid

            return None

    # --- Stopped ---

    def store_stopped_instance(self, app_id: str,

                               app_instance_id: str) -> None:

        if app_id not in self._stopped:

            self._stopped[app_id] = []

        self._stopped[app_id].append(app_instance_id)

        with self._lock:

            lst = self._stopped.setdefault(app_id, [])

            if app_instance_id not in lst:  # de-duplicate

                lst.append(app_instance_id)

    def get_stopped_instances(

        self,

            app_id: Optional[str] = None) -> List[str] | Dict[str, List[str]]:

        app_id: Optional[str] = None

    ) -> Union[List[str], Dict[str, List[str]]]:

        with self._lock:

            if app_id:

            return self._stopped.get(app_id, [])

        return self._stopped

                return list(self._stopped.get(app_id, []))

            return {aid: list(ids) for aid, ids in self._stopped.items()}

    def remove_stopped_instances(self, app_id: str) -> None:

        with self._lock:

            self._stopped.pop(app_id, None)