"""Defines the central class for the CubicWeb RQL server: the repository.

The repository is an abstraction allowing execution of rql queries against

data sources. Most of the work is actually done in helper classes. The

repository mainly:

* brings these classes all together to provide a single access

  point to a cubicweb instance.

* handles session management

* provides method for pyro registration, to call if pyro is enabled

:organization: Logilab

:copyright: 2001-2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.

:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses

"""

__docformat__ = "restructuredtext en"

import sys

import Queue

from os.path import join

from datetime import datetime

from time import time, localtime, strftime

from logilab.common.decorators import cached

from logilab.common.compat import any

from yams import BadSchemaDefinition

from rql import RQLSyntaxError

from cubicweb import (CW_SOFTWARE_ROOT, CW_MIGRATION_MAP,

                      UnknownEid, AuthenticationError, ExecutionError,

                      ETypeNotSupportedBySources, MultiSourcesError,

                      BadConnectionId, Unauthorized, ValidationError,

                      typed_eid)

from cubicweb import cwvreg, schema, server

from cubicweb.server import utils, hook, pool, querier, sources

from cubicweb.server.session import Session, InternalSession

class CleanupEidTypeCacheOp(hook.SingleLastOperation):

    """on rollback of a insert query or commit of delete query, we have to

    clear repository's cache from no more valid entries

    NOTE: querier's rqlst/solutions cache may have been polluted too with

    queries such as Any X WHERE X eid 32 if 32 has been rollbacked however

    generated queries are unpredictable and analysing all the cache probably

    too expensive. Notice that there is no pb when using args to specify eids

    instead of giving them into the rql string.

    """

    def commit_event(self):

        """the observed connections pool has been rollbacked,

        remove inserted eid from repository type/source cache

        """

        try:

            self.session.repo.clear_caches(

                self.session.transaction_data['pendingeids'])

        except KeyError:

            pass

    def rollback_event(self):

        """the observed connections pool has been rollbacked,

        remove inserted eid from repository type/source cache

        """

        try:

            self.session.repo.clear_caches(

                self.session.transaction_data['neweids'])

        except KeyError:

            pass

def del_existing_rel_if_needed(session, eidfrom, rtype, eidto):

    """delete existing relation when adding a new one if card is 1 or ?

    have to be done once the new relation has been inserted to avoid having

    an entity without a relation for some time

    this kind of behaviour has to be done in the repository so we don't have

    hooks order hazardness

    """

    # XXX now that rql in migraction default to unsafe_execute we don't want to

    #     skip that for super session (though we can still skip it for internal

    #     sessions). Also we should imo rely on the orm to first fetch existing

    #     entity if any then delete it.

    if session.is_internal_session \

           or not session.vreg.config.is_hook_category_activated('integrity'):

        return

    card = session.schema_rproperty(rtype, eidfrom, eidto, 'cardinality')

    # one may be tented to check for neweids but this may cause more than one

    # relation even with '1?'  cardinality if thoses relations are added in the

    # same transaction where the entity is being created. This never occurs from

    # the web interface but may occurs during test or dbapi connection (though

    # not expected for this).  So: don't do it, we pretend to ensure repository

    # consistency.

    #

    # XXX we don't want read permissions to be applied but we want delete

    # permission to be checked

    rschema = session.repo.schema.rschema(rtype)

    if card[0] in '1?':

        if not rschema.inlined: # inlined relations will be implicitly deleted

            rset = session.unsafe_execute('Any X,Y WHERE X %s Y, X eid %%(x)s, '

                                          'NOT Y eid %%(y)s' % rtype,

                                          {'x': eidfrom, 'y': eidto}, 'x')

            if rset:

                safe_delete_relation(session, rschema, *rset[0])

    if card[1] in '1?':

        rset = session.unsafe_execute('Any X,Y WHERE X %s Y, Y eid %%(y)s, '

                                      'NOT X eid %%(x)s' % rtype,

                                      {'x': eidfrom, 'y': eidto}, 'y')

        if rset:

            safe_delete_relation(session, rschema, *rset[0])

def safe_delete_relation(session, rschema, subject, object):

    if not rschema.has_perm(session, 'delete', fromeid=subject, toeid=object):

        raise Unauthorized()

    session.repo.glob_delete_relation(session, subject, rschema.type, object)

class Repository(object):

    """a repository provides access to a set of persistent storages for

    entities and relations

    XXX protect pyro access

    """

    def __init__(self, config, vreg=None, debug=False):

        self.config = config

        if vreg is None:

            vreg = cwvreg.CubicWebVRegistry(config, debug)

        self.vreg = vreg

        self.pyro_registered = False

        self.info('starting repository from %s', self.config.apphome)

        # dictionary of opened sessions

        self._sessions = {}

        # list of functions to be called at regular interval

        self._looping_tasks = []

        # list of running threads

        self._running_threads = []

        # initial schema, should be build or replaced latter

        self.schema = schema.CubicWebSchema(config.appid)

        self.vreg.schema = self.schema # until actual schema is loaded...

        # querier helper, need to be created after sources initialization

        self.querier = querier.QuerierHelper(self, self.schema)

        # sources

        self.sources = []

        self.sources_by_uri = {}

        # FIXME: store additional sources info in the system database ?

        # FIXME: sources should be ordered (add_entity priority)

        for uri, source_config in config.sources().items():

            if uri == 'admin':

                # not an actual source

                continue

            source = self.get_source(uri, source_config)

            self.sources_by_uri[uri] = source

            self.sources.append(source)

        self.system_source = self.sources_by_uri['system']

        # ensure system source is the first one

        self.sources.remove(self.system_source)

        self.sources.insert(0, self.system_source)

        # cache eid -> type / source

        self._type_source_cache = {}

        # cache (extid, source uri) -> eid

        self._extid_cache = {}

        # open some connections pools

        if config.open_connections_pools:

            self.open_connections_pools()

    def _boostrap_hook_registry(self):

        """called during bootstrap since we need the metadata hooks"""

        hooksdirectory = join(CW_SOFTWARE_ROOT, 'hooks')

        self.vreg.init_registration([hooksdirectory])

        self.vreg.load_file(join(hooksdirectory, 'metadata.py'),

                            'cubicweb.hooks.metadata')

    def open_connections_pools(self):

        config = self.config

        self._available_pools = Queue.Queue()

        self._available_pools.put_nowait(pool.ConnectionsPool(self.sources))

        if config.read_instance_schema:

            # normal start: load the instance schema from the database

            self.fill_schema()

        elif config.bootstrap_schema:

            # usually during repository creation

            self.warning("set fs instance'schema as bootstrap schema")

            config.bootstrap_cubes()

            self.set_schema(config.load_schema(), resetvreg=False)

            # need to load the Any and CWUser entity types

            etdirectory = join(CW_SOFTWARE_ROOT, 'entities')

            self.vreg.init_registration([etdirectory])

            for modname in ('__init__', 'authobjs', 'wfobjs'):

                self.vreg.load_file(join(etdirectory, '%s.py' % modname),

                                'cubicweb.entities.%s' % modname)

            self._boostrap_hook_registry()

        else:

            # test start: use the file system schema (quicker)

            self.warning("set fs instance'schema")

            config.bootstrap_cubes()

            self.set_schema(config.load_schema())

        if not config.creating:

            if 'CWProperty' in self.schema:

                self.vreg.init_properties(self.properties())

            # call source's init method to complete their initialisation if

            # needed (for instance looking for persistent configuration using an

            # internal session, which is not possible until pools have been

            # initialized)

            for source in self.sources:

                source.init()

        else:

            # call init_creating so for instance native source can configurate

            # tsearch according to postgres version

            for source in self.sources:

                source.init_creating()

        # close initialization pool and reopen fresh ones for proper

        # initialization now that we know cubes

        self._get_pool().close(True)

        # list of available pools (we can't iterated on Queue instance)

        self.pools = []

        for i in xrange(config['connections-pool-size']):

            self.pools.append(pool.ConnectionsPool(self.sources))

            self._available_pools.put_nowait(self.pools[-1])

        self._shutting_down = False

        self.hm = self.vreg['hooks']

        if not (config.creating or config.repairing):

            # call instance level initialisation hooks

            self.hm.call_hooks('server_startup', repo=self)

            # register a task to cleanup expired session

            self.looping_task(config['session-time']/3., self.clean_sessions)

    # internals ###############################################################

    def get_source(self, uri, source_config):

        source_config['uri'] = uri

        return sources.get_source(source_config, self.schema, self)

    def set_schema(self, schema, resetvreg=True, rebuildinfered=True):

        if rebuildinfered:

            schema.rebuild_infered_relations()

        self.info('set schema %s %#x', schema.name, id(schema))

        if resetvreg:

            if self.config._cubes is None:

                self.config.init_cubes(self.get_cubes())

            # full reload of all appobjects

            self.vreg.reset()

            self.vreg.set_schema(schema)

        else:

            self.vreg._set_schema(schema)

        self.querier.set_schema(schema)

        for source in self.sources:

            source.set_schema(schema)

        self.schema = schema

    def fill_schema(self):

        """lod schema from the repository"""

        from cubicweb.server.schemaserial import deserialize_schema

        self.info('loading schema from the repository')

        appschema = schema.CubicWebSchema(self.config.appid)

        self.set_schema(self.config.load_bootstrap_schema(), resetvreg=False)

        self.debug('deserializing db schema into %s %#x', appschema.name, id(appschema))

        session = self.internal_session()

        try:

            try:

                deserialize_schema(appschema, session)

            except BadSchemaDefinition:

                raise

            except Exception, ex:

                import traceback

                traceback.print_exc()

                raise Exception('Is the database initialised ? (cause: %s)' %

                                (ex.args and ex.args[0].strip() or 'unknown')), \

                                None, sys.exc_info()[-1]

        finally:

            session.close()

        self.set_schema(appschema)

    def start_looping_tasks(self):

        assert isinstance(self._looping_tasks, list), 'already started'

        for i, (interval, func, args) in enumerate(self._looping_tasks):

            self._looping_tasks[i] = task = utils.LoopTask(interval, func, args)

            self.info('starting task %s with interval %.2fs', task.name,

                      interval)

            task.start()

        # ensure no tasks will be further added

        self._looping_tasks = tuple(self._looping_tasks)

    def looping_task(self, interval, func, *args):

        """register a function to be called every `interval` seconds.

        looping tasks can only be registered during repository initialization,

        once done this method will fail.

        """

        try:

            self._looping_tasks.append( (interval, func, args) )

        except AttributeError:

            raise RuntimeError("can't add looping task once the repository is started")

    def threaded_task(self, func):

        """start function in a separated thread"""

        t = utils.RepoThread(func, self._running_threads)

        t.start()

    #@locked

    def _get_pool(self):

        try:

            return self._available_pools.get(True, timeout=5)

        except Queue.Empty:

            raise Exception('no pool available after 5 secs, probably either a '

                            'bug in code (too many uncommited/rollbacked '

                            'connections) or too much load on the server (in '

                            'which case you can try to set a bigger '

                            'connections pools size)')

    def _free_pool(self, pool):

        self._available_pools.put_nowait(pool)

    def pinfo(self):

        # XXX: session.pool is accessed from a local storage, would be interesting

        #      to see if there is a pool set in any thread specific data)

        import threading

        return '%s: %s (%s)' % (self._available_pools.qsize(),

                                ','.join(session.user.login for session in self._sessions.values()

                                         if session.pool),

                                threading.currentThread())

    def shutdown(self):

        """called on server stop event to properly close opened sessions and

        connections

        """

        assert not self._shutting_down, 'already shutting down'

        self._shutting_down = True

        if isinstance(self._looping_tasks, tuple): # if tasks have been started

            for looptask in self._looping_tasks:

                self.info('canceling task %s...', looptask.name)

                looptask.cancel()

                looptask.join()

                self.info('task %s finished', looptask.name)

        for thread in self._running_threads:

            self.info('waiting thread %s...', thread.name)

            thread.join()

            self.info('thread %s finished', thread.name)

        self.hm.call_hooks('server_shutdown', repo=self)

        self.close_sessions()

        while not self._available_pools.empty():

            pool = self._available_pools.get_nowait()

            try:

                pool.close(True)

            except:

                self.exception('error while closing %s' % pool)

                continue

        if self.pyro_registered:

            pyro_unregister(self.config)

        hits, misses = self.querier.cache_hit, self.querier.cache_miss

        try:

            self.info('rqlt st cache hit/miss: %s/%s (%s%% hits)', hits, misses,

                      (hits * 100) / (hits + misses))

            hits, misses = self.system_source.cache_hit, self.system_source.cache_miss

            self.info('sql cache hit/miss: %s/%s (%s%% hits)', hits, misses,

                      (hits * 100) / (hits + misses))

            nocache  = self.system_source.no_cache

            self.info('sql cache usage: %s/%s (%s%%)', hits+ misses, nocache,

                      ((hits + misses) * 100) / (hits + misses + nocache))

        except ZeroDivisionError:

            pass

    def stats(self): # XXX restrict to managers session?

        import threading

        results = {}

        for hits, misses, title in (

            (self.querier.cache_hit, self.querier.cache_miss, 'rqlt_st'),

            (self.system_source.cache_hit, self.system_source.cache_miss, 'sql'),

            ):

            results['%s_cache_hit' % title] =  hits

            results['%s_cache_miss' % title] = misses

            results['%s_cache_hit_percent' % title] = (hits * 100) / (hits + misses)

        results['sql_no_cache'] = self.system_source.no_cache

        results['nb_open_sessions'] = len(self._sessions)

        results['nb_active_threads'] = threading.activeCount()

        results['looping_tasks'] = ', '.join(str(t) for t in self._looping_tasks)