# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

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

#

# This file is part of CubicWeb.

#

# CubicWeb is free software: you can redistribute it and/or modify it under the

# terms of the GNU Lesser General Public License as published by the Free

# Software Foundation, either version 2.1 of the License, or (at your option)

# any later version.

#

# CubicWeb 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 Lesser General Public License for more

# details.

#

# You should have received a copy of the GNU Lesser General Public License along

# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

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

"""

from __future__ import with_statement

__docformat__ = "restructuredtext en"

import sys

import threading

import Queue

from warnings import warn

from itertools import chain

from os.path import join

from datetime import datetime

from time import time, localtime, strftime

from logilab.common.decorators import cached, clear_cache

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 rql.utils import rqlvar_maker

from cubicweb import (CW_SOFTWARE_ROOT, CW_MIGRATION_MAP, QueryError,

                      UnknownEid, AuthenticationError, ExecutionError,

                      ETypeNotSupportedBySources, MultiSourcesError,

                      BadConnectionId, Unauthorized, ValidationError,

                      RepositoryError, UniqueTogetherError, typed_eid, onevent)

from cubicweb import cwvreg, schema, server

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

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

     security_enabled

from cubicweb.server.ssplanner import EditedEntity

NO_CACHE_RELATIONS = set( [('owned_by', 'object'),

                           ('created_by', 'object'),

                           ('cw_source', 'object'),

                           ])

def prefill_entity_caches(entity, relations):

    session = entity._cw

    # prefill entity relation caches

    for rschema in entity.e_schema.subject_relations():

        rtype = str(rschema)

        if rtype in schema.VIRTUAL_RTYPES or (rtype, 'subject') in NO_CACHE_RELATIONS:

            continue

        if rschema.final:

            entity.cw_attr_cache.setdefault(rtype, None)

        else:

            entity.cw_set_relation_cache(rtype, 'subject',

                                         session.empty_rset())

    for rschema in entity.e_schema.object_relations():

        rtype = str(rschema)

        if rtype in schema.VIRTUAL_RTYPES or (rtype, 'object') in NO_CACHE_RELATIONS:

            continue

        entity.cw_set_relation_cache(rtype, 'object', session.empty_rset())

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 if integrity explicitly disabled

    if not session.is_hook_category_activated('activeintegrity'):

        return

    rdef = session.rtype_eids_rdef(rtype, eidfrom, eidto)

    card = rdef.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.

    #

    # notes:

    # * inlined relations will be implicitly deleted for the subject entity

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

    #   permission to be checked

    if card[0] in '1?':

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

    if card[1] in '1?':

        with security_enabled(session, read=False):

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

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

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

class NullEventBus(object):

    def publish(self, msg):

        pass

    def add_subscription(self, topic, callback):

        pass

    def start(self):

        pass

    def stop(self):

        pass

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

        self.config = config

        if vreg is None:

            vreg = cwvreg.CWRegistryStore(config)

        self.vreg = vreg

        self.pyro_registered = False

        self.pyro_uri = None

        self.app_instances_bus = NullEventBus()

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

        # shutdown flag

        self.shutting_down = False

        # sources (additional sources info in the system database)

        self.system_source = self.get_source('native', 'system',

                                             config.sources()['system'])

        self.sources = [self.system_source]

        self.sources_by_uri = {'system': self.system_source}

        # querier helper, need to be created after sources initialization

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

        # cache eid -> (type, physical source, extid, actual source)

        self._type_source_cache = {}

        # cache (extid, source uri) -> eid

        self._extid_cache = {}

        # open some connections set

        if config.init_cnxset_pool:

            self.init_cnxset_pool()

        @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 init_cnxset_pool(self):

        config = self.config

        self._cnxsets_pool = Queue.Queue()

        self._cnxsets_pool.put_nowait(pool.ConnectionsSet(self.sources))

        if config.quick_start:

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

            # information (eg dump/restore/...)

            config._cubes = ()

            # only load hooks and entity classes in the registry

            config.__class__.cube_appobject_path = set(('hooks', 'entities'))

            config.__class__.cubicweb_appobject_path = set(('hooks', 'entities'))

            self.set_schema(config.load_schema())

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

            # will be reinitialized later from cubes found in the database

            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)

            hooksdirectory = join(CW_SOFTWARE_ROOT, 'hooks')

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

                                'cubicweb.hooks.metadata')

        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:

            self.init_sources_from_database()

            if 'CWProperty' in self.schema:

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

        else:

            # call init_creating so that for instance native source can

            # configurate tsearch according to postgres version

            for source in self.sources:

                source.init_creating()

        # close initialization connetions set and reopen fresh ones for proper

        # initialization now that we know cubes

        self._get_cnxset().close(True)

        # list of available_cnxsets (we can't iterate on Queue instance)

        self.cnxsets = []

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

            self.cnxsets.append(pool.ConnectionsSet(self.sources))

            self._cnxsets_pool.put_nowait(self.cnxsets[-1])

        if config.quick_start:

            config.init_cubes(self.get_cubes())

        self.hm = hook.HooksManager(self.vreg)

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

    def init_sources_from_database(self):

        self.sources_by_eid = {}

        if self.config.quick_start \

               or not 'CWSource' in self.schema: # # 3.10 migration

            self.system_source.init_creating()

            return

        session = self.internal_session()

        try:

            # FIXME: sources should be ordered (add_entity priority)

            for sourceent in session.execute(

                'Any S, SN, SA, SC WHERE S is_instance_of CWSource, '

                'S name SN, S type SA, S config SC').entities():

                if sourceent.name == 'system':

                    self.system_source.eid = sourceent.eid

                    self.sources_by_eid[sourceent.eid] = self.system_source

                    self.system_source.init(True, sourceent)

                    continue

                self.add_source(sourceent, add_to_cnxsets=False)

        finally:

            session.close()

    def _clear_planning_caches(self):

        for cache in ('source_defs', 'is_multi_sources_relation',

                      'can_cross_relation', 'rel_type_sources'):

            clear_cache(self, cache)

    def add_source(self, sourceent, add_to_cnxsets=True):

        source = self.get_source(sourceent.type, sourceent.name,

                                 sourceent.host_config, sourceent.eid)

        self.sources_by_eid[sourceent.eid] = source

        self.sources_by_uri[sourceent.name] = source

        if self.config.source_enabled(source):

            # 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 connections sets have been

            # initialized)

            source.init(True, sourceent)

            if not source.copy_based_source:

                self.sources.append(source)

                self.querier.set_planner()

                if add_to_cnxsets:

                    for cnxset in self.cnxsets:

                       cnxset.add_source(source)

        else:

            source.init(False, sourceent)

        self._clear_planning_caches()

    def remove_source(self, uri):

        source = self.sources_by_uri.pop(uri)

        del self.sources_by_eid[source.eid]

        if self.config.source_enabled(source) and not source.copy_based_source:

            self.sources.remove(source)

            self.querier.set_planner()

            for cnxset in self.cnxsets:

                cnxset.remove_source(source)

        self._clear_planning_caches()

    def get_source(self, type, uri, source_config, eid=None):

        # set uri and type in source config so it's available through

        # source_defs()

        source_config['uri'] = uri

        source_config['type'] = type

        return sources.get_source(type, source_config, self, eid)

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

            # trigger full reload of all appobjects

            self.vreg.set_schema(schema)

        else:

            self.vreg._set_schema(schema)

        self.querier.set_schema(schema)

        # don't use self.sources, we may want to give schema even to disabled

        # sources

        for source in self.sources_by_uri.values():

            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.cleanup_session_time = self.config['cleanup-session-time'] or 60 * 60 * 24

            assert self.cleanup_session_time > 0

            cleanup_session_interval = min(60*60, self.cleanup_session_time / 3)

            self.looping_task(cleanup_session_interval, 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(self, 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) )