"""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

"""

from __future__ import with_statement

__docformat__ = "restructuredtext en"

import sys

import Queue

from os.path import join

from datetime import datetime

from time import time, localtime, strftime

#from pickle import dumps

from logilab.common.decorators import cached

from logilab.common.compat import any

from logilab.common import flatten

from yams import BadSchemaDefinition

from yams.schema import role_name

from rql import RQLSyntaxError

from cubicweb import (CW_SOFTWARE_ROOT, CW_MIGRATION_MAP,

                      UnknownEid, AuthenticationError, ExecutionError,

                      ETypeNotSupportedBySources, MultiSourcesError,

                      BadConnectionId, Unauthorized, ValidationError,

                      typed_eid, onevent)

from cubicweb import cwvreg, schema, server

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

from cubicweb.server.session import Session, InternalSession, InternalManager, \

     security_enabled

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

    """

    # skip that for internal session or if integrity explicitly disabled

    #

    # XXX 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.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

            with security_enabled(session, read=False):

                session.execute('DELETE X %s Y WHERE X eid %%(x)s, '

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

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

    if card[1] in '1?':

        with security_enabled(session, read=False):

            session.execute('DELETE X %sY WHERE Y eid %%(y)s, '

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

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

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()

        @onevent('after-registry-reload', self)

        def fix_user_classes(self):

            usercls = self.vreg['etypes'].etype_class('CWUser')

            for session in self._sessions.values():

                if not isinstance(session.user, InternalManager):

                    session.user.__class__ = usercls

    def _bootstrap_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.quick_start:

            # quick start, usually only to get a minimal repository to get cubes

            config._cubes = ()

            config['connections-pool-size'] = 1

            config._cubes = None

        elif config.creating:

            # repository creation

            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._bootstrap_hook_registry()

        elif config.read_instance_schema:

            # normal start: load the instance schema from the database

            self.fill_schema()

        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

        if config.quick_start:

            config.init_cubes(self.get_cubes())

    # 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):

        if not (self.config.creating or self.config.repairing

                or self.config.quick_start):

            # call instance level initialisation hooks

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

            # register a task to cleanup expired session

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

        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)

        if not (self.config.creating or self.config.repairing

                or self.config.quick_start):

            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 = {}

        querier = self.querier

        source = self.system_source

        for size, maxsize, hits, misses, title in (

            (len(querier._rql_cache), self.config['rql-cache-size'],

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

            (len(source._cache), self.config['rql-cache-size'],

            source.cache_hit, source.cache_miss, 'sql'),

            ):

            results['%s_cache_size' % title] =  '%s / %s' % (size, maxsize)

            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)

        results['available_pools'] = self._available_pools.qsize()