#!/usr/bin/env python3

# Libervia: an XMPP client
# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from collections import namedtuple
import copy
from dataclasses import dataclass
from functools import partial
import mimetypes
import os.path
from pathlib import Path
import time
from typing import Any, Dict, Optional, Tuple
from uuid import uuid4

import shortuuid
from twisted.internet import defer, error, reactor
from twisted.python import failure
from twisted.words.protocols.jabber import jid

from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.memory.crypto import BlockCipher
from libervia.backend.memory.crypto import PasswordHasher
from libervia.backend.memory.disco import Discovery
from libervia.backend.memory.params import Params
from libervia.backend.memory.persistent import PersistentDict
from libervia.backend.memory.sqla import (
    Notification,
    NotificationPriority,
    NotificationStatus,
    NotificationType,
    Storage,
)
from libervia.backend.tools import config as tools_config
from libervia.backend.tools.common import data_format
from libervia.backend.tools.common import regex


log = getLogger(__name__)


PresenceTuple = namedtuple("PresenceTuple", ("show", "priority", "statuses"))
MSG_NO_SESSION = "Session id doesn't exist or is finished"


class Sessions(object):
    """Sessions are data associated to key used for a temporary moment, with optional profile checking."""

    DEFAULT_TIMEOUT = 600

    def __init__(self, timeout=None, resettable_timeout=True):
        """
        @param timeout (int): nb of seconds before session destruction
        @param resettable_timeout (bool): if True, the timeout is reset on each access
        """
        self._sessions = dict()
        self.timeout = timeout or Sessions.DEFAULT_TIMEOUT
        self.resettable_timeout = resettable_timeout

    def new_session(self, session_data=None, session_id=None, profile=None):
        """Create a new session

        @param session_data: mutable data to use, default to a dict
        @param session_id (str): force the session_id to the given string
        @param profile: if set, the session is owned by the profile,
                        and profile_get must be used instead of __getitem__
        @return: session_id, session_data
        """
        if session_id is None:
            session_id = str(uuid4())
        elif session_id in self._sessions:
            raise exceptions.ConflictError(
                "Session id {} is already used".format(session_id)
            )
        timer = reactor.callLater(self.timeout, self._purge_session, session_id)
        if session_data is None:
            session_data = {}
        self._sessions[session_id] = (
            (timer, session_data) if profile is None else (timer, session_data, profile)
        )
        return session_id, session_data

    def _purge_session(self, session_id):
        try:
            timer, session_data, profile = self._sessions[session_id]
        except ValueError:
            timer, session_data = self._sessions[session_id]
            profile = None
        try:
            timer.cancel()
        except error.AlreadyCalled:
            # if the session is time-outed, the timer has been called
            pass
        del self._sessions[session_id]
        log.debug(
            "Session {} purged{}".format(
                session_id,
                " (profile {})".format(profile) if profile is not None else "",
            )
        )

    def __len__(self):
        return len(self._sessions)

    def __contains__(self, session_id):
        return session_id in self._sessions

    def profile_get(self, session_id, profile):
        try:
            timer, session_data, profile_set = self._sessions[session_id]
        except ValueError:
            raise exceptions.InternalError(
                "You need to use __getitem__ when profile is not set"
            )
        except KeyError:
            raise failure.Failure(KeyError(MSG_NO_SESSION))
        if profile_set != profile:
            raise exceptions.InternalError("current profile differ from set profile !")
        if self.resettable_timeout:
            timer.reset(self.timeout)
        return session_data

    def __getitem__(self, session_id):
        try:
            timer, session_data = self._sessions[session_id]
        except ValueError:
            raise exceptions.InternalError(
                "You need to use profile_get instead of __getitem__ when profile is set"
            )
        except KeyError:
            raise failure.Failure(KeyError(MSG_NO_SESSION))
        if self.resettable_timeout:
            timer.reset(self.timeout)
        return session_data

    def __setitem__(self, key, value):
        raise NotImplementedError("You need do use new_session to create a session")

    def __delitem__(self, session_id):
        """delete the session data"""
        self._purge_session(session_id)

    def keys(self):
        return list(self._sessions.keys())

    def iterkeys(self):
        return iter(self._sessions.keys())


class ProfileSessions(Sessions):
    """ProfileSessions extends the Sessions class, but here the profile can be
    used as the key to retrieve data or delete a session (instead of session id).
    """

    def _profile_get_all_ids(self, profile):
        """Return a list of the sessions ids that are associated to the given profile.

        @param profile: %(doc_profile)s
        @return: a list containing the sessions ids
        """
        ret = []
        for session_id in self._sessions.keys():
            try:
                timer, session_data, profile_set = self._sessions[session_id]
            except ValueError:
                continue
            if profile == profile_set:
                ret.append(session_id)
        return ret

    def profile_get_unique(self, profile):
        """Return the data of the unique session that is associated to the given profile.

        @param profile: %(doc_profile)s
        @return:
            - mutable data (default: dict) of the unique session
            - None if no session is associated to the profile
            - raise an error if more than one session are found
        """
        ids = self._profile_get_all_ids(profile)
        if len(ids) > 1:
            raise exceptions.InternalError(
                "profile_get_unique has been used but more than one session has been found!"
            )
        return (
            self.profile_get(ids[0], profile) if len(ids) == 1 else None
        )  # XXX: timeout might be reset

    def profile_del_unique(self, profile):
        """Delete the unique session that is associated to the given profile.

        @param profile: %(doc_profile)s
        @return: None, but raise an error if more than one session are found
        """
        ids = self._profile_get_all_ids(profile)
        if len(ids) > 1:
            raise exceptions.InternalError(
                "profile_del_unique has been used but more than one session has been found!"
            )
        if len(ids) == 1:
            del self._sessions[ids[0]]


class PasswordSessions(ProfileSessions):

    # FIXME: temporary hack for the user personal key not to be lost. The session
    # must actually be purged and later, when the personal key is needed, the
    # profile password should be asked again in order to decrypt it.
    def __init__(self, timeout=None):
        ProfileSessions.__init__(self, timeout, resettable_timeout=False)

    def _purge_session(self, session_id):
        log.debug(
            "FIXME: PasswordSessions should ask for the profile password after the session expired"
        )


class Memory:
    """This class manage all the persistent information"""

    def __init__(self, host):
        log.info(_("Memory manager init"))
        self.host = host
        self._entities_cache = {}  # XXX: keep presence/last resource/other data in cache
        #     /!\ an entity is not necessarily in roster
        #     main key is bare jid, value is a dict
        #     where main key is resource, or None for bare jid
        self._key_signals = set()  # key which need a signal to frontends when updated
        self.subscriptions = {}
        self.auth_sessions = PasswordSessions()  # remember the authenticated profiles
        self.disco = Discovery(host)
        self.config = tools_config.parse_main_conf(log_filenames=True)
        self._cache_path = Path(self.config_get("", "local_dir"), C.CACHE_DIR)
        self.admins = self.config_get("", "admins_list", [])
        self.admin_jids = set()
        self._file_path_lock = defer.DeferredLock()

    async def initialise(self):
        self.storage = Storage()
        await self.storage.initialise()
        PersistentDict.storage = self.storage
        self.params = Params(self.host, self.storage)
        log.info(_("Loading default params template"))
        self.params.load_default_params()
        await self.load()
        self.memory_data = PersistentDict("memory")
        await self.memory_data.load()
        await self.disco.load()
        for admin in self.admins:
            try:
                admin_jid_s = await self.param_get_a_async(
                    "JabberID", "Connection", profile_key=admin
                )
            except Exception as e:
                log.warning(f"Can't retrieve jid of admin {admin!r}: {e}")
            else:
                if admin_jid_s is not None:
                    try:
                        admin_jid = jid.JID(admin_jid_s).userhostJID()
                    except RuntimeError:
                        log.warning(f"Invalid JID for admin {admin}: {admin_jid_s}")
                    else:
                        self.admin_jids.add(admin_jid)

    ## Configuration ##

    def config_get(
        self, section: str | None, name: str, default: Any = None
    ) -> str | list | dict:
        """Get the main configuration option

        @param section: section of the config file (None or '' for DEFAULT)
        @param name: name of the option
        @param default: value to use if not found
        @return: str, list or dict
        """
        return tools_config.config_get(self.config, section, name, default)

    def load_xml(self, filename):
        """Load parameters template from xml file

        @param filename (str): input file
        @return: bool: True in case of success
        """
        if not filename:
            return False
        filename = os.path.expanduser(filename)
        if os.path.exists(filename):
            try:
                self.params.load_xml(filename)
                log.debug(_("Parameters loaded from file: %s") % filename)
                return True
            except Exception as e:
                log.error(_("Can't load parameters from file: %s") % e)
        return False

    def save_xml(self, filename):
        """Save parameters template to xml file

        @param filename (str): output file
        @return: bool: True in case of success
        """
        if not filename:
            return False
        # TODO: need to encrypt files (at least passwords !) and set permissions
        filename = os.path.expanduser(filename)
        try:
            self.params.save_xml(filename)
            log.debug(_("Parameters saved to file: %s") % filename)
            return True
        except Exception as e:
            log.error(_("Can't save parameters to file: %s") % e)
        return False

    def load(self):
        """Load parameters and all memory things from db"""
        # parameters data
        return self.params.load_gen_params()

    def load_individual_params(self, profile):
        """Load individual parameters for a profile
        @param profile: %(doc_profile)s"""
        return self.params.load_ind_params(profile)

    ## Profiles/Sessions management ##

    def start_session(self, password, profile):
        """ "Iniatialise session for a profile

        @param password(unicode): profile session password
            or empty string is no password is set
        @param profile: %(doc_profile)s
        @raise exceptions.ProfileUnknownError if profile doesn't exists
        @raise exceptions.PasswordError: the password does not match
        """
        profile = self.get_profile_name(profile)

        def create_session(__):
            """Called once params are loaded."""
            self._entities_cache[profile] = {}
            log.info("[{}] Profile session started".format(profile))
            return False

        def backend_initialised(__):
            def do_start_session(__=None):
                if self.is_session_started(profile):
                    log.info("Session already started!")
                    return True
                try:
                    # if there is a value at this point in self._entities_cache,
                    # it is the load_individual_params Deferred, the session is starting
                    session_d = self._entities_cache[profile]
                except KeyError:
                    # else we do request the params
                    session_d = self._entities_cache[profile] = (
                        self.load_individual_params(profile)
                    )
                    session_d.addCallback(create_session)
                finally:
                    return session_d

            auth_d = defer.ensureDeferred(self.profile_authenticate(password, profile))
            auth_d.addCallback(do_start_session)
            return auth_d

        if self.host.init_pre_script.called:
            return defer.succeed(None).addCallback(backend_initialised)
        else:
            return self.host.init_pre_script.addCallback(backend_initialised)

    def stop_session(self, profile):
        """Delete a profile session

        @param profile: %(doc_profile)s
        """
        if self.host.is_connected(profile):
            log.debug("Disconnecting profile because of session stop")
            self.host.disconnect(profile)
        self.auth_sessions.profile_del_unique(profile)
        try:
            self._entities_cache[profile]
        except KeyError:
            log.warning("Profile was not in cache")

    def _is_session_started(self, profile_key):
        return self.is_session_started(self.get_profile_name(profile_key))

    def is_session_started(self, profile):
        try:
            # XXX: if the value in self._entities_cache is a Deferred,
            #      the session is starting but not started yet
            return not isinstance(self._entities_cache[profile], defer.Deferred)
        except KeyError:
            return False

    async def profile_authenticate(self, password, profile):
        """Authenticate the profile.

        @param password (unicode): the SàT profile password
        @return: None in case of success (an exception is raised otherwise)
        @raise exceptions.PasswordError: the password does not match
        """
        if not password and self.auth_sessions.profile_get_unique(profile):
            # XXX: this allows any frontend to connect with the empty password as soon as
            # the profile has been authenticated at least once before. It is OK as long as
            # submitting a form with empty passwords is restricted to local frontends.
            return

        sat_cipher = await self.param_get_a_async(
            C.PROFILE_PASS_PATH[1], C.PROFILE_PASS_PATH[0], profile_key=profile
        )
        valid = PasswordHasher.verify(password, sat_cipher)
        if not valid:
            log.warning(
                _("Authentication failure of profile {profile}").format(profile=profile)
            )
            raise exceptions.PasswordError("The provided profile password doesn't match.")
        return await self.new_auth_session(password, profile)

    async def new_auth_session(self, key, profile):
        """Start a new session for the authenticated profile.

        If there is already an existing session, no new one is created
        The personal key is loaded encrypted from a PersistentDict before being decrypted.

        @param key: the key to decrypt the personal key
        @param profile: %(doc_profile)s
        """
        data = await PersistentDict(C.MEMORY_CRYPTO_NAMESPACE, profile).load()
        personal_key = BlockCipher.decrypt(key, data[C.MEMORY_CRYPTO_KEY])
        # Create the session for this profile and store the personal key
        session_data = self.auth_sessions.profile_get_unique(profile)
        if not session_data:
            self.auth_sessions.new_session(
                {C.MEMORY_CRYPTO_KEY: personal_key}, profile=profile
            )
            log.debug("auth session created for profile %s" % profile)

    def purge_profile_session(self, profile):
        """Delete cache of data of profile
        @param profile: %(doc_profile)s"""
        log.info(_("[%s] Profile session purge" % profile))
        self.params.purge_profile(profile)
        try:
            del self._entities_cache[profile]
        except KeyError:
            log.error(
                _("Trying to purge roster status cache for a profile not in memory: [%s]")
                % profile
            )

    def get_profiles_list(self, clients=True, components=False):
        """retrieve profiles list

        @param clients(bool): if True return clients profiles
        @param components(bool): if True return components profiles
        @return (list[unicode]): selected profiles
        """
        if not clients and not components:
            log.warning(_("requesting no profiles at all"))
            return []
        profiles = self.storage.get_profiles_list()
        if clients and components:
            return sorted(profiles)
        is_component = self.storage.profile_is_component
        if clients:
            p_filter = lambda p: not is_component(p)
        else:
            p_filter = lambda p: is_component(p)

        return sorted(p for p in profiles if p_filter(p))

    def get_profile_name(self, profile_key, return_profile_keys=False):
        """Return name of profile from keyword

        @param profile_key: can be the profile name or a keyword (like @DEFAULT@)
        @param return_profile_keys: if True, return unmanaged profile keys (like "@ALL@"). This keys must be managed by the caller
        @return: requested profile name
        @raise exceptions.ProfileUnknownError if profile doesn't exists
        """
        return self.params.get_profile_name(profile_key, return_profile_keys)

    def profile_set_default(self, profile):
        """Set default profile

        @param profile: %(doc_profile)s
        """
        # we want to be sure that the profile exists
        profile = self.get_profile_name(profile)

        self.memory_data["Profile_default"] = profile

    def create_profile(self, name, password, component=None):
        """Create a new profile

        @param name(unicode): profile name
        @param password(unicode): profile password
            Can be empty to disable password
        @param component(None, unicode): set to entry point if this is a component
        @return: Deferred
        @raise exceptions.NotFound: component is not a known plugin import name
        """
        if not name:
            raise ValueError("Empty profile name")
        if name[0] == "@":
            raise ValueError("A profile name can't start with a '@'")
        if "\n" in name:
            raise ValueError("A profile name can't contain line feed ('\\n')")

        if name in self._entities_cache:
            raise exceptions.ConflictError("A session for this profile exists")

        if component:
            if not component in self.host.plugins:
                raise exceptions.NotFound(
                    _(
                        "Can't find component {component} entry point".format(
                            component=component
                        )
                    )
                )
            # FIXME: PLUGIN_INFO is not currently accessible after import, but type shoul be tested here
            #  if self.host.plugins[component].PLUGIN_INFO[u"type"] != C.PLUG_TYPE_ENTRY_POINT:
            #      raise ValueError(_(u"Plugin {component} is not an entry point !".format(
            #          component = component)))

        d = self.params.create_profile(name, component)

        def init_personal_key(__):
            # be sure to call this after checking that the profile doesn't exist yet

            # generated once for all and saved in a PersistentDict
            personal_key = BlockCipher.get_random_key(base64=True).decode("utf-8")
            self.auth_sessions.new_session(
                {C.MEMORY_CRYPTO_KEY: personal_key}, profile=name
            )  # will be encrypted by param_set

        def start_fake_session(__):
            # avoid ProfileNotConnected exception in param_set
            self._entities_cache[name] = None
            self.params.load_ind_params(name)

        def stop_fake_session(__):
            del self._entities_cache[name]
            self.params.purge_profile(name)

        d.addCallback(init_personal_key)
        d.addCallback(start_fake_session)
        d.addCallback(
            lambda __: self.param_set(
                C.PROFILE_PASS_PATH[1], password, C.PROFILE_PASS_PATH[0], profile_key=name
            )
        )
        d.addCallback(stop_fake_session)
        d.addCallback(lambda __: self.auth_sessions.profile_del_unique(name))
        return d

    def profile_delete_async(self, name, force=False):
        """Delete an existing profile

        @param name: Name of the profile
        @param force: force the deletion even if the profile is connected.
        To be used for direct calls only (not through the bridge).
        @return: a Deferred instance
        """

        def clean_memory(__):
            self.auth_sessions.profile_del_unique(name)
            try:
                del self._entities_cache[name]
            except KeyError:
                pass

        d = self.params.profile_delete_async(name, force)
        d.addCallback(clean_memory)
        return d

    def is_component(self, profile_name):
        """Tell if a profile is a component

        @param profile_name(unicode): name of the profile
        @return (bool): True if profile is a component
        @raise exceptions.NotFound: profile doesn't exist
        """
        return self.storage.profile_is_component(profile_name)

    def get_entry_point(self, profile_name):
        """Get a component entry point

        @param profile_name(unicode): name of the profile
        @return (bool): True if profile is a component
        @raise exceptions.NotFound: profile doesn't exist
        """
        return self.storage.get_entry_point(profile_name)

    ## History ##

    def add_to_history(self, client, data):
        return self.storage.add_to_history(data, client.profile)

    def _history_get_serialise(self, history_data):
        return [
            (
                uid,
                timestamp,
                from_jid,
                to_jid,
                message,
                subject,
                mess_type,
                data_format.serialise(extra),
            )
            for uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra in history_data
        ]

    def _history_get(
        self,
        from_jid_s,
        to_jid_s,
        limit=C.HISTORY_LIMIT_NONE,
        between=True,
        filters=None,
        profile=C.PROF_KEY_NONE,
    ):
        from_jid = jid.JID(from_jid_s) if from_jid_s else None
        to_jid = jid.JID(to_jid_s) if to_jid_s else None
        d = self.history_get(from_jid, to_jid, limit, between, filters, profile)
        d.addCallback(self._history_get_serialise)
        return d

    def history_get(
        self,
        from_jid: jid.JID | None,
        to_jid: jid.JID | None,
        limit: int = C.HISTORY_LIMIT_NONE,
        between: bool = True,
        filters: dict[str, str] | None = None,
        profile: str = C.PROF_KEY_NONE,
    ) -> defer.Deferred[list]:
        """Retrieve messages in history

        @param from_jid: source JID (full, or bare for catchall)
        @param to_jid: dest JID (full, or bare for catchall)
        @param limit: maximum number of messages to get:
            - 0 for no message (returns the empty list)
            - C.HISTORY_LIMIT_NONE or None for unlimited
            - C.HISTORY_LIMIT_DEFAULT to use the HISTORY_LIMIT parameter value
        @param between: confound source and dest (ignore the direction)
        @param filters: pattern to filter the history results
            (see bridge API for details)
        @param profile: %(doc_profile)s
        @return: list of message data as in [message_new]
        """
        assert profile != C.PROF_KEY_NONE
        if limit == C.HISTORY_LIMIT_DEFAULT:
            limit = int(self.param_get_a(C.HISTORY_LIMIT, "General", profile_key=profile))
        elif limit == C.HISTORY_LIMIT_NONE:
            limit = None
        if limit == 0:
            return defer.succeed([])
        return self.storage.history_get(
            from_jid, to_jid, limit, between, filters, profile
        )

    ## Statuses ##

    def _get_presence_statuses(self, profile_key):
        ret = self.presence_statuses_get(profile_key)
        return {entity.full(): data for entity, data in ret.items()}

    def presence_statuses_get(self, profile_key):
        """Get all the presence statuses of a profile

        @param profile_key: %(doc_profile_key)s
        @return: presence data: key=entity JID, value=presence data for this entity
        """
        client = self.host.get_client(profile_key)
        profile_cache = self._get_profile_cache(client)
        entities_presence = {}

        for entity_jid, entity_data in profile_cache.items():
            for resource, resource_data in entity_data.items():
                full_jid = copy.copy(entity_jid)
                full_jid.resource = resource
                try:
                    presence_data = self.get_entity_datum(client, full_jid, "presence")
                except KeyError:
                    continue
                entities_presence.setdefault(entity_jid, {})[
                    resource or ""
                ] = presence_data

        return entities_presence

    def set_presence_status(self, entity_jid, show, priority, statuses, profile_key):
        """Change the presence status of an entity

        @param entity_jid: jid.JID of the entity
        @param show: show status
        @param priority: priority
        @param statuses: dictionary of statuses
        @param profile_key: %(doc_profile_key)s
        """
        client = self.host.get_client(profile_key)
        presence_data = PresenceTuple(show, priority, statuses)
        self.update_entity_data(client, entity_jid, "presence", presence_data)
        if entity_jid.resource and show != C.PRESENCE_UNAVAILABLE:
            # If a resource is available, bare jid should not have presence information
            try:
                self.del_entity_datum(client, entity_jid.userhostJID(), "presence")
            except (KeyError, exceptions.UnknownEntityError):
                pass

    ## Resources ##

    def _get_all_resource(self, jid_s, profile_key):
        client = self.host.get_client(profile_key)
        jid_ = jid.JID(jid_s)
        return self.get_all_resources(client, jid_)

    def get_all_resources(self, client, entity_jid):
        """Return all resource from jid for which we have had data in this session

        @param entity_jid: bare jid of the entity
        return (set[unicode]): set of resources

        @raise exceptions.UnknownEntityError: if entity is not in cache
        @raise ValueError: entity_jid has a resource
        """
        # FIXME: is there a need to keep cache data for resources which are not connected anymore?
        if entity_jid.resource:
            raise ValueError(
                "get_all_resources must be used with a bare jid (got {})".format(
                    entity_jid
                )
            )
        profile_cache = self._get_profile_cache(client)
        try:
            entity_data = profile_cache[entity_jid.userhostJID()]
        except KeyError:
            raise exceptions.UnknownEntityError(
                "Entity {} not in cache".format(entity_jid)
            )
        resources = set(entity_data.keys())
        resources.discard(None)
        return resources

    def get_available_resources(self, client, entity_jid):
        """Return available resource for entity_jid

        This method differs from get_all_resources by returning only available resources
        @param entity_jid: bare jid of the entit
        return (list[unicode]): list of available resources

        @raise exceptions.UnknownEntityError: if entity is not in cache
        """
        available = []
        for resource in self.get_all_resources(client, entity_jid):
            full_jid = copy.copy(entity_jid)
            full_jid.resource = resource
            try:
                presence_data = self.get_entity_datum(client, full_jid, "presence")
            except KeyError:
                log.debug("Can't get presence data for {}".format(full_jid))
            else:
                if presence_data.show != C.PRESENCE_UNAVAILABLE:
                    available.append(resource)
        return available

    def _get_main_resource(self, jid_s, profile_key):
        client = self.host.get_client(profile_key)
        jid_ = jid.JID(jid_s)
        return self.main_resource_get(client, jid_) or ""

    def main_resource_get(self, client, entity_jid):
        """Return the main resource used by an entity

        @param entity_jid: bare entity jid
        @return (unicode): main resource or None
        """
        if entity_jid.resource:
            raise ValueError(
                "main_resource_get must be used with a bare jid (got {})".format(
                    entity_jid
                )
            )
        try:
            if self.host.plugins["XEP-0045"].is_joined_room(client, entity_jid):
                return None  # MUC rooms have no main resource
        except KeyError:  # plugin not found
            pass
        try:
            resources = self.get_all_resources(client, entity_jid)
        except exceptions.UnknownEntityError:
            log.warning("Entity is not in cache, we can't find any resource")
            return None
        priority_resources = []
        for resource in resources:
            full_jid = copy.copy(entity_jid)
            full_jid.resource = resource
            try:
                presence_data = self.get_entity_datum(client, full_jid, "presence")
            except KeyError:
                log.debug("No presence information for {}".format(full_jid))
                continue
            priority_resources.append((resource, presence_data.priority))
        try:
            return max(priority_resources, key=lambda res_tuple: res_tuple[1])[0]
        except ValueError:
            log.warning("No resource found at all for {}".format(entity_jid))
            return None

    ## Entities data ##

    def _get_profile_cache(self, client):
        """Check profile validity and return its cache

        @param client: SatXMPPClient
        @return (dict): profile cache
        """
        return self._entities_cache[client.profile]

    def set_signal_on_update(self, key, signal=True):
        """Set a signal flag on the key

        When the key will be updated, a signal will be sent to frontends
        @param key: key to signal
        @param signal(boolean): if True, do the signal
        """
        if signal:
            self._key_signals.add(key)
        else:
            self._key_signals.discard(key)

    def get_all_entities_iter(self, client, with_bare=False):
        """Return an iterator of full jids of all entities in cache

        @param with_bare: if True, include bare jids
        @return (list[unicode]): list of jids
        """
        profile_cache = self._get_profile_cache(client)
        # we construct a list of all known full jids (bare jid of entities x resources)
        for bare_jid, entity_data in profile_cache.items():
            for resource in entity_data.keys():
                if resource is None:
                    continue
                full_jid = copy.copy(bare_jid)
                full_jid.resource = resource
                yield full_jid

    def update_entity_data(self, client, entity_jid, key, value, silent=False):
        """Set a misc data for an entity

        If key was registered with set_signal_on_update, a signal will be sent to frontends
        @param entity_jid: JID of the entity, C.ENTITY_ALL_RESOURCES for all resources of
            all entities, C.ENTITY_ALL for all entities (all resources + bare jids)
        @param key: key to set (eg: C.ENTITY_TYPE)
        @param value: value for this key (eg: C.ENTITY_TYPE_MUC)
        @param silent(bool): if True, doesn't send signal to frontend, even if there is a
            signal flag (see set_signal_on_update)
        """
        profile_cache = self._get_profile_cache(client)
        if entity_jid in (C.ENTITY_ALL_RESOURCES, C.ENTITY_ALL):
            entities = self.get_all_entities_iter(client, entity_jid == C.ENTITY_ALL)
        else:
            entities = (entity_jid,)

        for jid_ in entities:
            entity_data = profile_cache.setdefault(jid_.userhostJID(), {}).setdefault(
                jid_.resource, {}
            )

            entity_data[key] = value
            if key in self._key_signals and not silent:
                self.host.bridge.entity_data_updated(
                    jid_.full(), key, data_format.serialise(value), client.profile
                )

    def del_entity_datum(self, client, entity_jid, key):
        """Delete a data for an entity

        @param entity_jid: JID of the entity, C.ENTITY_ALL_RESOURCES for all resources of all entities,
                           C.ENTITY_ALL for all entities (all resources + bare jids)
        @param key: key to delete (eg: C.ENTITY_TYPE)

        @raise exceptions.UnknownEntityError: if entity is not in cache
        @raise KeyError: key is not in cache
        """
        profile_cache = self._get_profile_cache(client)
        if entity_jid in (C.ENTITY_ALL_RESOURCES, C.ENTITY_ALL):
            entities = self.get_all_entities_iter(client, entity_jid == C.ENTITY_ALL)
        else:
            entities = (entity_jid,)

        for jid_ in entities:
            try:
                entity_data = profile_cache[jid_.userhostJID()][jid_.resource]
            except KeyError:
                raise exceptions.UnknownEntityError("Entity {} not in cache".format(jid_))
            try:
                del entity_data[key]
            except KeyError as e:
                if entity_jid in (C.ENTITY_ALL_RESOURCES, C.ENTITY_ALL):
                    continue  # we ignore KeyError when deleting keys from several entities
                else:
                    raise e

    def _get_entities_data(self, entities_jids, keys_list, profile_key):
        client = self.host.get_client(profile_key)
        ret = self.entities_data_get(
            client, [jid.JID(jid_) for jid_ in entities_jids], keys_list
        )
        return {
            jid_.full(): {k: data_format.serialise(v) for k, v in data.items()}
            for jid_, data in ret.items()
        }

    def entities_data_get(self, client, entities_jids, keys_list=None):
        """Get a list of cached values for several entities at once

        @param entities_jids: jids of the entities, or empty list for all entities in cache
        @param keys_list (iterable,None): list of keys to get, None for everything
        @param profile_key: %(doc_profile_key)s
        @return: dict withs values for each key in keys_list.
                 if there is no value of a given key, resulting dict will
                 have nothing with that key nether
                 if an entity doesn't exist in cache, it will not appear
                 in resulting dict

        @raise exceptions.UnknownEntityError: if entity is not in cache
        """

        def fill_entity_data(entity_cache_data):
            entity_data = {}
            if keys_list is None:
                entity_data = entity_cache_data
            else:
                for key in keys_list:
                    try:
                        entity_data[key] = entity_cache_data[key]
                    except KeyError:
                        continue
            return entity_data

        profile_cache = self._get_profile_cache(client)
        ret_data = {}
        if entities_jids:
            for entity in entities_jids:
                try:
                    entity_cache_data = profile_cache[entity.userhostJID()][
                        entity.resource
                    ]
                except KeyError:
                    continue
                ret_data[entity.full()] = fill_entity_data(entity_cache_data, keys_list)
        else:
            for bare_jid, data in profile_cache.items():
                for resource, entity_cache_data in data.items():
                    full_jid = copy.copy(bare_jid)
                    full_jid.resource = resource
                    ret_data[full_jid] = fill_entity_data(entity_cache_data)

        return ret_data

    def _get_entity_data(self, entity_jid_s, keys_list=None, profile=C.PROF_KEY_NONE):
        return self.entity_data_get(
            self.host.get_client(profile), jid.JID(entity_jid_s), keys_list
        )

    def entity_data_get(self, client, entity_jid, keys_list=None):
        """Get a list of cached values for entity

        @param entity_jid: JID of the entity
        @param keys_list (iterable,None): list of keys to get, None for everything
        @param profile_key: %(doc_profile_key)s
        @return: dict withs values for each key in keys_list.
                 if there is no value of a given key, resulting dict will
                 have nothing with that key nether

        @raise exceptions.UnknownEntityError: if entity is not in cache
        """
        profile_cache = self._get_profile_cache(client)
        try:
            entity_data = profile_cache[entity_jid.userhostJID()][entity_jid.resource]
        except KeyError:
            raise exceptions.UnknownEntityError(
                "Entity {} not in cache (was requesting {})".format(entity_jid, keys_list)
            )
        if keys_list is None:
            return entity_data

        return {key: entity_data[key] for key in keys_list if key in entity_data}

    def get_entity_datum(self, client, entity_jid, key):
        """Get a datum from entity

        @param entity_jid: JID of the entity
        @param key: key to get
        @return: requested value

        @raise exceptions.UnknownEntityError: if entity is not in cache
        @raise KeyError: if there is no value for this key and this entity
        """
        return self.entity_data_get(client, entity_jid, (key,))[key]

    def del_entity_cache(
        self, entity_jid, delete_all_resources=True, profile_key=C.PROF_KEY_NONE
    ):
        """Remove all cached data for entity

        @param entity_jid: JID of the entity to delete
        @param delete_all_resources: if True also delete all known resources from cache (a bare jid must be given in this case)
        @param profile_key: %(doc_profile_key)s

        @raise exceptions.UnknownEntityError: if entity is not in cache
        """
        client = self.host.get_client(profile_key)
        profile_cache = self._get_profile_cache(client)

        if delete_all_resources:
            if entity_jid.resource:
                raise ValueError(_("Need a bare jid to delete all resources"))
            try:
                del profile_cache[entity_jid]
            except KeyError:
                raise exceptions.UnknownEntityError(
                    "Entity {} not in cache".format(entity_jid)
                )
        else:
            try:
                del profile_cache[entity_jid.userhostJID()][entity_jid.resource]
            except KeyError:
                raise exceptions.UnknownEntityError(
                    "Entity {} not in cache".format(entity_jid)
                )

    ## Encryption ##

    def encrypt_value(self, value, profile):
        """Encrypt a value for the given profile. The personal key must be loaded
        already in the profile session, that should be the case if the profile is
        already authenticated.

        @param value (str): the value to encrypt
        @param profile (str): %(doc_profile)s
        @return: the deferred encrypted value
        """
        try:
            personal_key = self.auth_sessions.profile_get_unique(profile)[
                C.MEMORY_CRYPTO_KEY
            ]
        except TypeError:
            raise exceptions.InternalError(
                _("Trying to encrypt a value for %s while the personal key is undefined!")
                % profile
            )
        return BlockCipher.encrypt(personal_key, value)

    def decrypt_value(self, value, profile):
        """Decrypt a value for the given profile. The personal key must be loaded
        already in the profile session, that should be the case if the profile is
        already authenticated.

        @param value (str): the value to decrypt
        @param profile (str): %(doc_profile)s
        @return: the deferred decrypted value
        """
        try:
            personal_key = self.auth_sessions.profile_get_unique(profile)[
                C.MEMORY_CRYPTO_KEY
            ]
        except TypeError:
            raise exceptions.InternalError(
                _("Trying to decrypt a value for %s while the personal key is undefined!")
                % profile
            )
        return BlockCipher.decrypt(personal_key, value)

    def encrypt_personal_data(self, data_key, data_value, crypto_key, profile):
        """Re-encrypt a personal data (saved to a PersistentDict).

        @param data_key: key for the individual PersistentDict instance
        @param data_value: the value to be encrypted
        @param crypto_key: the key to encrypt the value
        @param profile: %(profile_doc)s
        @return: a deferred None value
        """

        def got_ind_memory(data):
            data[data_key] = BlockCipher.encrypt(crypto_key, data_value)
            return data.force(data_key)

        def done(__):
            log.debug(
                _("Personal data (%(ns)s, %(key)s) has been successfuly encrypted")
                % {"ns": C.MEMORY_CRYPTO_NAMESPACE, "key": data_key}
            )

        d = PersistentDict(C.MEMORY_CRYPTO_NAMESPACE, profile).load()
        return d.addCallback(got_ind_memory).addCallback(done)

    ## Subscription requests ##

    def add_waiting_sub(self, type_, entity_jid, profile_key):
        """Called when a subcription request is received"""
        profile = self.get_profile_name(profile_key)
        assert profile
        if profile not in self.subscriptions:
            self.subscriptions[profile] = {}
        self.subscriptions[profile][entity_jid] = type_

    def del_waiting_sub(self, entity_jid, profile_key):
        """Called when a subcription request is finished"""
        profile = self.get_profile_name(profile_key)
        assert profile
        if profile in self.subscriptions and entity_jid in self.subscriptions[profile]:
            del self.subscriptions[profile][entity_jid]

    def sub_waiting_get(self, profile_key):
        """Called to get a list of currently waiting subscription requests"""
        profile = self.get_profile_name(profile_key)
        if not profile:
            log.error(_("Asking waiting subscriptions for a non-existant profile"))
            return {}
        if profile not in self.subscriptions:
            return {}

        return self.subscriptions[profile]

    ## Parameters ##

    def get_string_param_a(
        self, name, category, attr="value", profile_key=C.PROF_KEY_NONE
    ):
        return self.params.get_string_param_a(name, category, attr, profile_key)

    def param_get_a(self, name, category, attr="value", profile_key=C.PROF_KEY_NONE):
        return self.params.param_get_a(name, category, attr, profile_key=profile_key)

    def param_get_a_async(
        self,
        name,
        category,
        attr="value",
        security_limit=C.NO_SECURITY_LIMIT,
        profile_key=C.PROF_KEY_NONE,
    ):
        return self.params.param_get_a_async(
            name, category, attr, security_limit, profile_key
        )

    def _get_params_values_from_category(
        self, category, security_limit, app, extra_s, profile_key
    ):
        return self.params._get_params_values_from_category(
            category, security_limit, app, extra_s, profile_key
        )

    def async_get_string_param_a(
        self,
        name,
        category,
        attribute="value",
        security_limit=C.NO_SECURITY_LIMIT,
        profile_key=C.PROF_KEY_NONE,
    ):

        profile = self.get_profile_name(profile_key)
        return defer.ensureDeferred(
            self.params.async_get_string_param_a(
                name, category, attribute, security_limit, profile
            )
        )

    def _get_params_ui(self, security_limit, app, extra_s, profile_key):
        return self.params._get_params_ui(security_limit, app, extra_s, profile_key)

    def params_categories_get(self):
        return self.params.params_categories_get()

    def param_set(
        self,
        name,
        value,
        category,
        security_limit=C.NO_SECURITY_LIMIT,
        profile_key=C.PROF_KEY_NONE,
    ):
        return self.params.param_set(name, value, category, security_limit, profile_key)

    def update_params(self, xml):
        return self.params.update_params(xml)

    def params_register_app(self, xml, security_limit=C.NO_SECURITY_LIMIT, app=""):
        return self.params.params_register_app(xml, security_limit, app)

    def set_default(self, name, category, callback, errback=None):
        return self.params.set_default(name, category, callback, errback)

    ## Private Data ##

    def _private_data_set(self, namespace, key, data_s, profile_key):
        client = self.host.get_client(profile_key)
        # we accept any type
        data = data_format.deserialise(data_s, type_check=None)
        return defer.ensureDeferred(
            self.storage.set_private_value(
                namespace, key, data, binary=True, profile=client.profile
            )
        )

    def _private_data_get(self, namespace, key, profile_key):
        client = self.host.get_client(profile_key)
        d = defer.ensureDeferred(
            self.storage.get_privates(
                namespace, [key], binary=True, profile=client.profile
            )
        )
        d.addCallback(lambda data_dict: data_format.serialise(data_dict.get(key)))
        return d

    def _private_data_delete(self, namespace, key, profile_key):
        client = self.host.get_client(profile_key)
        return defer.ensureDeferred(
            self.storage.del_private_value(
                namespace, key, binary=True, profile=client.profile
            )
        )

    ## Files ##

    def check_file_permission(
        self,
        file_data: dict,
        peer_jid: Optional[jid.JID],
        perms_to_check: Optional[Tuple[str]],
        set_affiliation: bool = False,
    ) -> None:
        """Check that an entity has the right permission on a file

        @param file_data: data of one file, as returned by get_files
        @param peer_jid: entity trying to access the file
        @param perms_to_check: permissions to check
            tuple of C.ACCESS_PERM_*
        @param check_parents: if True, also check all parents until root node
        @parma set_affiliation: if True, "affiliation" metadata will be set
        @raise exceptions.PermissionError: peer_jid doesn't have all permission
            in perms_to_check for file_data
        @raise exceptions.InternalError: perms_to_check is invalid
        """
        # TODO: knowing if user is owner is not enough, we need to check permission
        #   to see if user can modify/delete files, and set corresponding affiliation (publisher, member)
        if peer_jid is None and perms_to_check is None:
            return
        peer_jid = peer_jid.userhostJID()
        if peer_jid == file_data["owner"]:
            if set_affiliation:
                file_data["affiliation"] = "owner"
            # the owner has all rights, nothing to check
            return
        if not C.ACCESS_PERMS.issuperset(perms_to_check):
            raise exceptions.InternalError(_("invalid permission"))

        for perm in perms_to_check:
            # we check each perm and raise PermissionError as soon as one condition is not valid
            # we must never return here, we only return after the loop if nothing was blocking the access
            try:
                perm_data = file_data["access"][perm]
                perm_type = perm_data["type"]
            except KeyError:
                # No permission is set.
                # If we are in a root file/directory, we deny access
                # otherwise, we use public permission, as the parent directory will
                # block anyway, this avoid to have to recursively change permissions for
                # all sub directories/files when modifying a permission
                if not file_data.get("parent"):
                    raise exceptions.PermissionError()
                else:
                    perm_type = C.ACCESS_TYPE_PUBLIC
            if perm_type == C.ACCESS_TYPE_PUBLIC:
                continue
            elif perm_type == C.ACCESS_TYPE_WHITELIST:
                try:
                    jids = perm_data["jids"]
                except KeyError:
                    raise exceptions.PermissionError()
                if peer_jid.full() in jids:
                    continue
                else:
                    raise exceptions.PermissionError()
            else:
                raise exceptions.InternalError(
                    _("unknown access type: {type}").format(type=perm_type)
                )

    async def check_permission_to_root(self, client, file_data, peer_jid, perms_to_check):
        """do check_file_permission on file_data and all its parents until root"""
        current = file_data
        while True:
            self.check_file_permission(current, peer_jid, perms_to_check)
            parent = current["parent"]
            if not parent:
                break
            files_data = await self.get_files(
                client, peer_jid=None, file_id=parent, perms_to_check=None
            )
            try:
                current = files_data[0]
            except IndexError:
                raise exceptions.DataError("Missing parent")

    async def _get_parent_dir(
        self, client, path, parent, namespace, owner, peer_jid, perms_to_check
    ):
        """Retrieve parent node from a path, or last existing directory

        each directory of the path will be retrieved, until the last existing one
        @return (tuple[unicode, list[unicode])): parent, remaining path elements:
            - parent is the id of the last retrieved directory (or u'' for root)
            - remaining path elements are the directories which have not been retrieved
              (i.e. which don't exist)
        """
        # if path is set, we have to retrieve parent directory of the file(s) from it
        if parent is not None:
            raise exceptions.ConflictError(
                _("You can't use path and parent at the same time")
            )
        path_elts = [_f for _f in path.split("/") if _f]
        if {"..", "."}.intersection(path_elts):
            raise ValueError(_('".." or "." can\'t be used in path'))

        # we retrieve all directories from path until we get the parent container
        # non existing directories will be created
        parent = ""
        for idx, path_elt in enumerate(path_elts):
            directories = await self.storage.get_files(
                client,
                parent=parent,
                type_=C.FILE_TYPE_DIRECTORY,
                name=path_elt,
                namespace=namespace,
                owner=owner,
            )
            if not directories:
                return (parent, path_elts[idx:])
                # from this point, directories don't exist anymore, we have to create them
            elif len(directories) > 1:
                raise exceptions.InternalError(
                    _("Several directories found, this should not happen")
                )
            else:
                directory = directories[0]
                self.check_file_permission(directory, peer_jid, perms_to_check)
                parent = directory["id"]
        return (parent, [])

    def get_file_affiliations(self, file_data: dict) -> Dict[jid.JID, str]:
        """Convert file access to pubsub like affiliations"""
        affiliations = {}
        access_data = file_data["access"]

        read_data = access_data.get(C.ACCESS_PERM_READ, {})
        if read_data.get("type") == C.ACCESS_TYPE_WHITELIST:
            for entity_jid_s in read_data["jids"]:
                entity_jid = jid.JID(entity_jid_s)
                affiliations[entity_jid] = "member"

        write_data = access_data.get(C.ACCESS_PERM_WRITE, {})
        if write_data.get("type") == C.ACCESS_TYPE_WHITELIST:
            for entity_jid_s in write_data["jids"]:
                entity_jid = jid.JID(entity_jid_s)
                affiliations[entity_jid] = "publisher"

        owner = file_data.get("owner")
        if owner:
            affiliations[owner] = "owner"

        return affiliations

    def _set_file_affiliations_update(
        self, access: dict, file_data: dict, affiliations: Dict[jid.JID, str]
    ) -> None:
        read_data = access.setdefault(C.ACCESS_PERM_READ, {})
        if read_data.get("type") != C.ACCESS_TYPE_WHITELIST:
            read_data["type"] = C.ACCESS_TYPE_WHITELIST
            if "jids" not in read_data:
                read_data["jids"] = []
        read_whitelist = read_data["jids"]
        write_data = access.setdefault(C.ACCESS_PERM_WRITE, {})
        if write_data.get("type") != C.ACCESS_TYPE_WHITELIST:
            write_data["type"] = C.ACCESS_TYPE_WHITELIST
            if "jids" not in write_data:
                write_data["jids"] = []
        write_whitelist = write_data["jids"]
        for entity_jid, affiliation in affiliations.items():
            entity_jid_s = entity_jid.full()
            if affiliation == "none":
                try:
                    read_whitelist.remove(entity_jid_s)
                except ValueError:
                    log.warning(
                        "removing affiliation from an entity without read permission: "
                        f"{entity_jid}"
                    )
                try:
                    write_whitelist.remove(entity_jid_s)
                except ValueError:
                    pass
            elif affiliation == "publisher":
                if entity_jid_s not in read_whitelist:
                    read_whitelist.append(entity_jid_s)
                if entity_jid_s not in write_whitelist:
                    write_whitelist.append(entity_jid_s)
            elif affiliation == "member":
                if entity_jid_s not in read_whitelist:
                    read_whitelist.append(entity_jid_s)
                try:
                    write_whitelist.remove(entity_jid_s)
                except ValueError:
                    pass
            elif affiliation == "owner":
                raise NotImplementedError('"owner" affiliation can\'t be set')
            else:
                raise ValueError(f"unknown affiliation: {affiliation!r}")

    async def set_file_affiliations(
        self, client, file_data: dict, affiliations: Dict[jid.JID, str]
    ) -> None:
        """Apply pubsub like affiliation to file_data

        Affiliations are converted to access types, then set in a whitelist.
        Affiliation are mapped as follow:
            - "owner" can't be set (for now)
            - "publisher" gives read and write permissions
            - "member" gives read permission only
            - "none" removes both read and write permissions
        """
        file_id = file_data["id"]
        await self.file_update(
            file_id,
            "access",
            update_cb=partial(
                self._set_file_affiliations_update,
                file_data=file_data,
                affiliations=affiliations,
            ),
        )

    def _set_file_access_model_update(
        self, access: dict, file_data: dict, access_model: str
    ) -> None:
        read_data = access.setdefault(C.ACCESS_PERM_READ, {})
        if access_model == "open":
            requested_type = C.ACCESS_TYPE_PUBLIC
        elif access_model == "whitelist":
            requested_type = C.ACCESS_TYPE_WHITELIST
        else:
            raise ValueError(f"unknown access model: {access_model}")

        read_data["type"] = requested_type
        if requested_type == C.ACCESS_TYPE_WHITELIST and "jids" not in read_data:
            read_data["jids"] = []

    async def set_file_access_model(
        self,
        client,
        file_data: dict,
        access_model: str,
    ) -> None:
        """Apply pubsub like access_model to file_data

        Only 2 access models are supported so far:
            - "open": set public access to file/dir
            - "whitelist": set whitelist to file/dir
        """
        file_id = file_data["id"]
        await self.file_update(
            file_id,
            "access",
            update_cb=partial(
                self._set_file_access_model_update,
                file_data=file_data,
                access_model=access_model,
            ),
        )

    def get_files_owner(
        self,
        client,
        owner: Optional[jid.JID],
        peer_jid: Optional[jid.JID],
        file_id: Optional[str] = None,
        parent: Optional[str] = None,
    ) -> jid.JID:
        """Get owner to use for a file operation

        if owner is not explicitely set, a suitable one will be used (client.jid for
        clients, peer_jid for components).
        @raise exception.InternalError: we are one a component, and neither owner nor
            peer_jid are set
        """
        if owner is not None:
            return owner.userhostJID()
        if client is None:
            # client may be None when looking for file with public_id
            return None
        if file_id or parent:
            # owner has already been filtered on parent file
            return None
        if not client.is_component:
            return client.jid.userhostJID()
        if peer_jid is None:
            raise exceptions.InternalError(
                "Owner must be set for component if peer_jid is None"
            )
        return peer_jid.userhostJID()

    async def get_files(
        self,
        client,
        peer_jid,
        file_id=None,
        version=None,
        parent=None,
        path=None,
        type_=None,
        file_hash=None,
        hash_algo=None,
        name=None,
        namespace=None,
        mime_type=None,
        public_id=None,
        owner=None,
        access=None,
        projection=None,
        unique=False,
        perms_to_check=(C.ACCESS_PERM_READ,),
    ):
        """Retrieve files with with given filters

        @param peer_jid(jid.JID, None): jid trying to access the file
            needed to check permission.
            Use None to ignore permission (perms_to_check must be None too)
        @param file_id(unicode, None): id of the file
            None to ignore
        @param version(unicode, None): version of the file
            None to ignore
            empty string to look for current version
        @param parent(unicode, None): id of the directory containing the files
            None to ignore
            empty string to look for root files/directories
        @param path(Path, unicode, None): path to the directory containing the files
        @param type_(unicode, None): type of file filter, can be one of C.FILE_TYPE_*
        @param file_hash(unicode, None): hash of the file to retrieve
        @param hash_algo(unicode, None): algorithm use for file_hash
        @param name(unicode, None): name of the file to retrieve
        @param namespace(unicode, None): namespace of the files to retrieve
        @param mime_type(unicode, None): filter on this mime type
        @param public_id(unicode, None): filter on this public id
        @param owner(jid.JID, None): if not None, only get files from this owner
        @param access(dict, None): get file with given access (see [set_file])
        @param projection(list[unicode], None): name of columns to retrieve
            None to retrieve all
        @param unique(bool): if True will remove duplicates
        @param perms_to_check(tuple[unicode],None): permission to check
            must be a tuple of C.ACCESS_PERM_* or None
            if None, permission will no be checked (peer_jid must be None too in this
            case)
        other params are the same as for [set_file]
        @return (list[dict]): files corresponding to filters
        @raise exceptions.NotFound: parent directory not found (when path is specified)
        @raise exceptions.PermissionError: peer_jid can't use perms_to_check for one of
                                           the file
            on the path
        """
        if peer_jid is None and perms_to_check or perms_to_check is None and peer_jid:
            raise exceptions.InternalError(
                "if you want to disable permission check, both peer_jid and "
                "perms_to_check must be None"
            )
        owner = self.get_files_owner(client, owner, peer_jid, file_id, parent)
        if path is not None:
            path = str(path)
            # permission are checked by _get_parent_dir
            parent, remaining_path_elts = await self._get_parent_dir(
                client, path, parent, namespace, owner, peer_jid, perms_to_check
            )
            if remaining_path_elts:
                # if we have remaining path elements,
                # the parent directory is not found
                raise failure.Failure(exceptions.NotFound())
        if parent and peer_jid:
            # if parent is given directly and permission check is requested,
            # we need to check all the parents
            parent_data = await self.storage.get_files(client, file_id=parent)
            try:
                parent_data = parent_data[0]
            except IndexError:
                raise exceptions.DataError("mising parent")
            await self.check_permission_to_root(
                client, parent_data, peer_jid, perms_to_check
            )

        files = await self.storage.get_files(
            client,
            file_id=file_id,
            version=version,
            parent=parent,
            type_=type_,
            file_hash=file_hash,
            hash_algo=hash_algo,
            name=name,
            namespace=namespace,
            mime_type=mime_type,
            public_id=public_id,
            owner=owner,
            access=access,
            projection=projection,
            unique=unique,
        )

        if peer_jid:
            # if permission are checked, we must remove all file that user can't access
            to_remove = []
            for file_data in files:
                try:
                    self.check_file_permission(
                        file_data, peer_jid, perms_to_check, set_affiliation=True
                    )
                except exceptions.PermissionError:
                    to_remove.append(file_data)
            for file_data in to_remove:
                files.remove(file_data)
        return files

    async def set_file(
        self,
        client,
        name,
        file_id=None,
        version="",
        parent=None,
        path=None,
        type_=C.FILE_TYPE_FILE,
        file_hash=None,
        hash_algo=None,
        size=None,
        namespace=None,
        mime_type=None,
        public_id=None,
        created=None,
        modified=None,
        owner=None,
        access=None,
        extra=None,
        peer_jid=None,
        perms_to_check=(C.ACCESS_PERM_WRITE,),
    ):
        """Set a file metadata

        @param name(unicode): basename of the file
        @param file_id(unicode): unique id of the file
        @param version(unicode): version of this file
            empty string for current version or when there is no versioning
        @param parent(unicode, None): id of the directory containing the files
        @param path(unicode, None): virtual path of the file in the namespace
            if set, parent must be None. All intermediate directories will be created
            if needed, using current access.
        @param type_(str, None): type of file filter, can be one of C.FILE_TYPE_*
        @param file_hash(unicode): unique hash of the payload
        @param hash_algo(unicode): algorithm used for hashing the file (usually sha-256)
        @param size(int): size in bytes
        @param namespace(unicode, None): identifier (human readable is better) to group
                                         files
            For instance, namespace could be used to group files in a specific photo album
        @param mime_type(unicode): MIME type of the file, or None if not known/guessed
        @param public_id(unicode): id used to share publicly the file via HTTP
        @param created(int): UNIX time of creation
        @param modified(int,None): UNIX time of last modification, or None to use
                                   created date
        @param owner(jid.JID, None): jid of the owner of the file (mainly useful for
                                     component)
            will be used to check permission (only bare jid is used, don't use with MUC).
            Use None to ignore permission (perms_to_check must be None too)
        @param access(dict, None): serialisable dictionary with access rules.
            None (or empty dict) to use private access, i.e. allow only profile's jid to
            access the file
            key can be on on C.ACCESS_PERM_*,
            then a sub dictionary with a type key is used (one of C.ACCESS_TYPE_*).
            According to type, extra keys can be used:
                - C.ACCESS_TYPE_PUBLIC: the permission is granted for everybody
                - C.ACCESS_TYPE_WHITELIST: the permission is granted for jids (as unicode)
                  in the 'jids' key
            will be encoded to json in database
        @param extra(dict, None): serialisable dictionary of any extra data
            will be encoded to json in database
        @param perms_to_check(tuple[unicode],None): permission to check
            must be a tuple of C.ACCESS_PERM_* or None
            if None, permission will not be checked (peer_jid must be None too in this
            case)
        @param profile(unicode): profile owning the file
        """
        if "/" in name:
            raise ValueError('name must not contain a slash ("/")')
        if file_id is None:
            file_id = shortuuid.uuid()
        if (
            file_hash is not None
            and hash_algo is None
            or hash_algo is not None
            and file_hash is None
        ):
            raise ValueError("file_hash and hash_algo must be set at the same time")
        if mime_type is None:
            mime_type, __ = mimetypes.guess_type(name)
        else:
            mime_type = mime_type.lower()
        if public_id is not None:
            assert len(public_id) > 0
        if created is None:
            created = time.time()
        if namespace is not None:
            namespace = namespace.strip() or None
        if type_ == C.FILE_TYPE_DIRECTORY:
            if any((version, file_hash, size, mime_type)):
                raise ValueError(
                    "version, file_hash, size and mime_type can't be set for a directory"
                )
        owner = self.get_files_owner(client, owner, peer_jid, file_id, parent)

        if path is not None:
            # We use a lock to avoid race condition leading to duplicate directories with
            # same name and parent.
            await self._file_path_lock.acquire()
            try:

                path = str(path)
                # _get_parent_dir will check permissions if peer_jid is set, so we use owner
                parent, remaining_path_elts = await self._get_parent_dir(
                    client, path, parent, namespace, owner, owner, perms_to_check
                )
                # if remaining directories don't exist, we have to create them
                for new_dir in remaining_path_elts:
                    new_dir_id = shortuuid.uuid()
                    await self.storage.set_file(
                        client,
                        name=new_dir,
                        file_id=new_dir_id,
                        version="",
                        parent=parent,
                        type_=C.FILE_TYPE_DIRECTORY,
                        namespace=namespace,
                        created=time.time(),
                        owner=owner,
                        access=access,
                        extra={},
                    )
                    parent = new_dir_id
            finally:
                self._file_path_lock.release()

        elif parent is None:
            parent = ""

        await self.storage.set_file(
            client,
            file_id=file_id,
            version=version,
            parent=parent,
            type_=type_,
            file_hash=file_hash,
            hash_algo=hash_algo,
            name=name,
            size=size,
            namespace=namespace,
            mime_type=mime_type,
            public_id=public_id,
            created=created,
            modified=modified,
            owner=owner,
            access=access,
            extra=extra,
        )

    async def file_get_used_space(
        self, client, peer_jid: jid.JID, owner: Optional[jid.JID] = None
    ) -> int:
        """Get space taken by all files owned by an entity

        @param peer_jid: entity requesting the size
        @param owner: entity owning the file to check. If None, will be determined by
            get_files_owner
        @return: size of total space used by files of this owner
        """
        owner = self.get_files_owner(client, owner, peer_jid)
        if peer_jid.userhostJID() != owner and client.profile not in self.admins:
            raise exceptions.PermissionError("You are not allowed to check this size")
        return await self.storage.file_get_used_space(client, owner)

    def file_update(self, file_id, column, update_cb):
        """Update a file column taking care of race condition

        access is NOT checked in this method, it must be checked beforehand
        @param file_id(unicode): id of the file to update
        @param column(unicode): one of "access" or "extra"
        @param update_cb(callable): method to update the value of the colum
            the method will take older value as argument, and must update it in place
            Note that the callable must be thread-safe
        """
        return self.storage.file_update(file_id, column, update_cb)

    @defer.inlineCallbacks
    def _delete_file(
        self,
        client,
        peer_jid: jid.JID,
        recursive: bool,
        files_path: Path,
        file_data: dict,
    ):
        """Internal method to delete files/directories recursively

        @param peer_jid(jid.JID): entity requesting the deletion (must be owner of files
            to delete)
        @param recursive(boolean): True if recursive deletion is needed
        @param files_path(unicode): path of the directory containing the actual files
        @param file_data(dict): data of the file to delete
        """
        if file_data["owner"] != peer_jid:
            raise exceptions.PermissionError(
                "file {file_name} can't be deleted, {peer_jid} is not the owner".format(
                    file_name=file_data["name"], peer_jid=peer_jid.full()
                )
            )
        if file_data["type"] == C.FILE_TYPE_DIRECTORY:
            sub_files = yield self.get_files(client, peer_jid, parent=file_data["id"])
            if sub_files and not recursive:
                raise exceptions.DataError(_("Can't delete directory, it is not empty"))
            # we first delete the sub-files
            for sub_file_data in sub_files:
                if sub_file_data["type"] == C.FILE_TYPE_DIRECTORY:
                    sub_file_path = files_path / sub_file_data["name"]
                else:
                    sub_file_path = files_path
                yield self._delete_file(
                    client, peer_jid, recursive, sub_file_path, sub_file_data
                )
            # then the directory itself
            yield self.storage.file_delete(file_data["id"])
        elif file_data["type"] == C.FILE_TYPE_FILE:
            log.info(
                _("deleting file {name} with hash {file_hash}").format(
                    name=file_data["name"], file_hash=file_data["file_hash"]
                )
            )
            yield self.storage.file_delete(file_data["id"])
            references = yield self.get_files(
                client, peer_jid, file_hash=file_data["file_hash"]
            )
            if references:
                log.debug("there are still references to the file, we keep it")
            else:
                file_path = os.path.join(files_path, file_data["file_hash"])
                log.info(
                    _("no reference left to {file_path}, deleting").format(
                        file_path=file_path
                    )
                )
                try:
                    os.unlink(file_path)
                except FileNotFoundError:
                    log.error(
                        f"file at {file_path!r} doesn't exist but it was referenced in files database"
                    )
        else:
            raise exceptions.InternalError(
                "Unexpected file type: {file_type}".format(file_type=file_data["type"])
            )

    async def file_delete(self, client, peer_jid, file_id, recursive=False):
        """Delete a single file or a directory and all its sub-files

        @param file_id(unicode): id of the file to delete
        @param peer_jid(jid.JID): entity requesting the deletion,
            must be owner of all files to delete
        @param recursive(boolean): must be True to delete a directory and all sub-files
        """
        # FIXME: we only allow owner of file to delete files for now, but WRITE access
        #        should be checked too
        files_data = await self.get_files(client, peer_jid, file_id)
        if not files_data:
            raise exceptions.NotFound(
                "Can't find the file with id {file_id}".format(file_id=file_id)
            )
        file_data = files_data[0]
        if file_data["type"] != C.FILE_TYPE_DIRECTORY and recursive:
            raise ValueError("recursive can only be set for directories")
        files_path = self.host.get_local_path(None, C.FILES_DIR)
        await self._delete_file(client, peer_jid, recursive, files_path, file_data)

    ## Cache ##

    def get_cache_path(self, namespace: str, *args: str) -> Path:
        """Get path to use to get a common path for a namespace

        This can be used by plugins to manage permanent data. It's the responsability
        of plugins to clean this directory from unused data.
        @param namespace: unique namespace to use
        @param args: extra identifier which will be added to the path
        """
        namespace = namespace.strip().lower()
        return Path(
            self._cache_path,
            regex.path_escape(namespace),
            *(regex.path_escape(a) for a in args),
        )

    ## Notifications ##

    def _add_notification(
        self,
        type_: str,
        body_plain: str,
        body_rich: str,
        title: str,
        is_global: bool,
        requires_action: bool,
        priority: str,
        expire_at: float,
        extra_s: str,
        profile_key: str,
    ) -> defer.Deferred:
        client = self.host.get_client(profile_key)

        if not client.is_admin:
            raise exceptions.PermissionError("Only admins can add a notification")

        try:
            notification_type = NotificationType[type_]
            notification_priority = NotificationPriority[priority]
        except KeyError as e:
            raise exceptions.DataError(f"invalid notification type or priority data: {e}")

        return defer.ensureDeferred(
            self.add_notification(
                client,
                notification_type,
                body_plain,
                body_rich or None,
                title or None,
                is_global,
                requires_action,
                notification_priority,
                expire_at or None,
                data_format.deserialise(extra_s),
            )
        )

    async def add_notification(
        self,
        client: SatXMPPEntity,
        type_: NotificationType,
        body_plain: str,
        body_rich: Optional[str] = None,
        title: Optional[str] = None,
        is_global: bool = False,
        requires_action: bool = False,
        priority: NotificationPriority = NotificationPriority.MEDIUM,
        expire_at: Optional[float] = None,
        extra: Optional[dict] = None,
    ) -> None:
        """Create and broadcast a new notification.

        @param client: client associated with the notification. If None, the notification
            will be global (i.e. for all profiles).
        @param type_: type of the notification.
        @param body_plain: plain text body.
        @param body_rich: rich text (XHTML) body.
        @param title: optional title.
        @param is_global: True if the notification is for all profiles.
        @param requires_action: True if the notification requires user action (e.g. a
            dialog need to be answered).
        @priority: how urgent the notification is
        @param expire_at: expiration timestamp for the notification.
        @param extra: additional data.
        """
        notification = await self.storage.add_notification(
            None if is_global else client,
            type_,
            body_plain,
            body_rich,
            title,
            requires_action,
            priority,
            expire_at,
            extra,
        )
        self.host.bridge.notification_new(
            str(notification.id),
            notification.timestamp,
            type_.value,
            body_plain,
            body_rich or "",
            title or "",
            requires_action,
            priority.value,
            expire_at or 0,
            data_format.serialise(extra) if extra else "",
            C.PROF_KEY_ALL if is_global else client.profile,
        )

    def _get_notifications(self, filters_s: str, profile_key: str) -> defer.Deferred:
        """Fetch notifications for bridge with given filters and profile key.

        @param filters_s: serialized filter conditions. Keys can be:
            :type_ (str):
                Filter by type of the notification.
            :status (str):
                Filter by status of the notification.
            :requires_action (bool):
                Filter by notifications that require user action.
            :min_priority (str):
                Filter by minimum priority value.
        @param profile_key: key of the profile to fetch notifications for.
        @return: Deferred which fires with a list of serialised notifications.
        """
        client = self.host.get_client(profile_key)

        filters = data_format.deserialise(filters_s)

        try:
            if "type" in filters:
                filters["type_"] = NotificationType[filters.pop("type")]
            if "status" in filters:
                filters["status"] = NotificationStatus[filters["status"]]
            if "min_priority" in filters:
                filters["min_priority"] = NotificationPriority[
                    filters["min_priority"]
                ].value
        except KeyError as e:
            raise exceptions.DataError(f"invalid filter data: {e}")

        d = defer.ensureDeferred(self.storage.get_notifications(client, **filters))
        d.addCallback(
            lambda notifications: data_format.serialise(
                [notification.serialise() for notification in notifications]
            )
        )
        return d

    def _delete_notification(
        self, id_: str, is_global: bool, profile_key: str
    ) -> defer.Deferred:
        client = self.host.get_client(profile_key)
        if is_global and not client.is_admin:
            raise exceptions.PermissionError(
                "Only admins can delete global notifications"
            )
        return defer.ensureDeferred(self.delete_notification(client, id_, is_global))

    async def delete_notification(
        self, client: SatXMPPEntity, id_: str, is_global: bool = False
    ) -> None:
        """Delete a notification

        the notification must be from the requesting profile.
        @param id_: ID of the notification
        is_global: if True, a global notification will be removed.
        """
        await self.storage.delete_notification(None if is_global else client, id_)
        self.host.bridge.notification_deleted(
            id_, C.PROF_KEY_ALL if is_global else client.profile
        )

    def _notifications_expired_clean(
        self, limit_timestamp: float, profile_key: str
    ) -> defer.Deferred:
        if profile_key == C.PROF_KEY_NONE:
            client = None
        else:
            client = self.host.get_client(profile_key)

        return defer.ensureDeferred(
            self.storage.clean_expired_notifications(
                client, None if limit_timestamp == -1.0 else limit_timestamp
            )
        )

    ## Misc ##

    def is_entity_available(self, client, entity_jid):
        """Tell from the presence information if the given entity is available.

        @param entity_jid (JID): the entity to check (if bare jid is used, all resources are tested)
        @return (bool): True if entity is available
        """
        if not entity_jid.resource:
            return bool(
                self.get_available_resources(client, entity_jid)
            )  # is any resource is available, entity is available
        try:
            presence_data = self.get_entity_datum(client, entity_jid, "presence")
        except KeyError:
            log.debug("No presence information for {}".format(entity_jid))
            return False
        return presence_data.show != C.PRESENCE_UNAVAILABLE

    def is_admin(self, profile: str) -> bool:
        """Tell if given profile has administrator privileges"""
        return profile in self.admins

    def is_admin_jid(self, entity: jid.JID) -> bool:
        """Tells if an entity jid correspond to an admin one

        It is sometime not possible to use the profile alone to check if an entity is an
        admin (e.g. a request managed by a component). In this case we check if the JID
        correspond to an admin profile
        """
        return entity.userhostJID() in self.admin_jids