/*
 * Copyright 2014 The Apache Software Foundation.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.usergrid.corepersistence;


import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService;
import org.apache.usergrid.corepersistence.index.CollectionSettingsFactory;
import org.apache.usergrid.corepersistence.pipeline.read.ResultsPage;
import org.apache.usergrid.corepersistence.results.ConnectionRefQueryExecutor;
import org.apache.usergrid.corepersistence.results.EntityQueryExecutor;
import org.apache.usergrid.corepersistence.results.IdQueryExecutor;
import org.apache.usergrid.corepersistence.service.CollectionSearch;
import org.apache.usergrid.corepersistence.service.CollectionService;
import org.apache.usergrid.corepersistence.service.ConnectionSearch;
import org.apache.usergrid.corepersistence.service.ConnectionService;
import org.apache.usergrid.corepersistence.util.CpCollectionUtils;
import org.apache.usergrid.corepersistence.util.CpEntityMapUtils;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.*;
import org.apache.usergrid.persistence.Query.Level;
import org.apache.usergrid.persistence.cassandra.ConnectionRefImpl;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.entities.Group;
import org.apache.usergrid.persistence.entities.User;
import org.apache.usergrid.persistence.graph.*;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdge;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchEdgeType;
import org.apache.usergrid.persistence.index.query.Identifier;
import org.apache.usergrid.persistence.map.MapManager;
import org.apache.usergrid.persistence.map.MapScope;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.queue.settings.IndexConsistency;
import org.apache.usergrid.persistence.queue.settings.QueueIndexingStrategy;
import org.apache.usergrid.persistence.schema.CollectionInfo;
import org.apache.usergrid.utils.InflectionUtils;
import org.apache.usergrid.utils.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import rx.Observable;

import java.util.*;

import static org.apache.usergrid.corepersistence.util.CpNamingUtils.*;
import static org.apache.usergrid.persistence.Schema.*;
import static org.apache.usergrid.utils.InflectionUtils.singularize;
import static org.apache.usergrid.utils.MapUtils.addMapSet;


/**
 * Implement good-old Usergrid RelationManager with the new-fangled Core Persistence API.
 */
public class CpRelationManager implements RelationManager {

    private static final Logger logger = LoggerFactory.getLogger( CpRelationManager.class );
    private final EntityManagerFig entityManagerFig;

    private ManagerCache managerCache;

    private EntityManager em;

    private UUID applicationId;

    private EntityRef headEntity;

    private org.apache.usergrid.persistence.model.entity.Entity cpHeadEntity;

    private final ApplicationScope applicationScope;

    private final AsyncEventService indexService;

    private final CollectionSettingsFactory collectionSettingsFactory;


    private final CollectionService collectionService;
    private final ConnectionService connectionService;


    public CpRelationManager( final ManagerCache managerCache,
                              final AsyncEventService indexService, final CollectionService collectionService,
                              final ConnectionService connectionService,
                              final EntityManager em,
                              final EntityManagerFig entityManagerFig, final UUID applicationId,
                              final CollectionSettingsFactory collectionSettingsFactory,
                              final EntityRef headEntity) {


        Assert.notNull( em, "Entity manager cannot be null" );
        Assert.notNull( applicationId, "Application Id cannot be null" );
        Assert.notNull( headEntity, "Head entity cannot be null" );
        Assert.notNull( headEntity.getUuid(), "Head entity uuid cannot be null" );
        Assert.notNull( indexService, "indexService cannot be null" );
        Assert.notNull( collectionService, "collectionService cannot be null" );
        Assert.notNull( connectionService, "connectionService cannot be null" );

        this.entityManagerFig = entityManagerFig;

        // TODO: this assert should not be failing
        //Assert.notNull( indexBucketLocator, "indexBucketLocator cannot be null" );
        this.em = em;
        this.applicationId = applicationId;
        this.headEntity = headEntity;
        this.managerCache = managerCache;
        this.applicationScope = CpNamingUtils.getApplicationScope( applicationId );

        this.collectionService = collectionService;
        this.connectionService = connectionService;

        if ( logger.isDebugEnabled() ) {
            logger.debug( "Loading head entity {}:{} from app {}",
                headEntity.getType(), headEntity.getUuid(), applicationScope
            );
        }

        Id entityId = new SimpleId( headEntity.getUuid(), headEntity.getType() );

        this.cpHeadEntity = ( ( CpEntityManager ) em ).load( entityId );

        // commented out because it is possible that CP entity has not been created yet
        Assert.notNull( cpHeadEntity, String
            .format( "cpHeadEntity cannot be null for entity id %s, app id %s", entityId.getUuid(), applicationId ) );

        this.indexService = indexService;
        this.collectionSettingsFactory = collectionSettingsFactory;

    }


    @Override
    public Set<String> getCollectionIndexes( String collectionName ) throws Exception {
        GraphManager gm = managerCache.getGraphManager( applicationScope );

        String edgeTypePrefix = CpNamingUtils.getEdgeTypeFromCollectionName( collectionName );

        if (logger.isTraceEnabled()) {
            logger.trace("getCollectionIndexes(): Searching for edge type prefix {} to target {}:{}",
                edgeTypePrefix, cpHeadEntity.getId().getType(), cpHeadEntity.getId().getUuid() );
        }

        Observable<Set<String>> types =
            gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( cpHeadEntity.getId(), edgeTypePrefix, null ) )
              .collect( () -> new HashSet<>(), ( set, type ) -> set.add( type ) );


        return types.toBlocking().last();
    }


    @Override
    public Map<String, Map<UUID, Set<String>>> getOwners() throws Exception {

        // TODO: do we need to restrict this to edges prefixed with owns?
        //Map<EntityRef, Set<String>> containerEntities = getContainers(-1, "owns", null);
        Map<EntityRef, Set<String>> containerEntities = getContainers();

        Map<String, Map<UUID, Set<String>>> owners = new LinkedHashMap<String, Map<UUID, Set<String>>>();

        for ( EntityRef owner : containerEntities.keySet() ) {
            Set<String> collections = containerEntities.get( owner );
            for ( String collection : collections ) {
                MapUtils.addMapMapSet( owners, owner.getType(), owner.getUuid(), collection );
            }
        }

        return owners;
    }


    private Map<EntityRef, Set<String>> getContainers() {
        return getContainers( -1, null, null );
    }


    /**
     * Gets containing collections and/or connections depending on the edge type you pass in
     *
     * @param limit Max number to return
     * @param edgeType Edge type, edge type prefix or null to allow any edge type
     * @param fromEntityType Only consider edges from entities of this type
     */
    Map<EntityRef, Set<String>> getContainers( final int limit, final String edgeType, final String fromEntityType ) {

        final GraphManager gm = managerCache.getGraphManager( applicationScope );

        Observable<Edge> edges =
            gm.getEdgeTypesToTarget( new SimpleSearchEdgeType( cpHeadEntity.getId(), edgeType, null ) )
              .flatMap( edgeType1 -> gm.loadEdgesToTarget(
                  new SimpleSearchByEdgeType( cpHeadEntity.getId(), edgeType1, Long.MAX_VALUE,
                      SearchByEdgeType.Order.DESCENDING, Optional.<Edge>absent() ) ) );

        //if our limit is set, take them.  Note this logic is still borked, we can't possibly fit everything in memmory
        if ( limit > -1 ) {
            edges = edges.take( limit );
        }


        return edges.collect( () -> new LinkedHashMap<EntityRef, Set<String>>(), ( entityRefSetMap, edge ) -> {
            if ( fromEntityType != null && !fromEntityType.equals( edge.getSourceNode().getType() ) ) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Ignoring edge from entity type {}", edge.getSourceNode().getType());
                }
                return;
            }

            final EntityRef eref =
                new SimpleEntityRef( edge.getSourceNode().getType(), edge.getSourceNode().getUuid() );

            String name = getNameFromEdgeType( edge.getType() );
            addMapSet( entityRefSetMap, eref, name );
        } ).toBlocking().last();
    }


    @Override
    public boolean isConnectionMember( String connectionType, EntityRef entity ) throws Exception {

        Id entityId = new SimpleId( entity.getUuid(), entity.getType() );


        if (logger.isTraceEnabled()) {
            logger.trace("isConnectionMember(): Checking for edge type {} from {}:{} to {}:{}",
                connectionType, headEntity.getType(), headEntity.getUuid(), entity.getType(), entity.getUuid() );
        }

        GraphManager gm = managerCache.getGraphManager( applicationScope );
        Observable<MarkedEdge> edges = gm.loadEdgeVersions( CpNamingUtils
            .createEdgeFromConnectionType( new SimpleId( headEntity.getUuid(), headEntity.getType() ), connectionType,
                entityId ) );

        return edges.toBlocking().firstOrDefault( null ) != null;
    }


    @SuppressWarnings( "unchecked" )
    @Override
    public boolean isCollectionMember( String collectionName, EntityRef entity ) throws Exception {

        Id entityId = new SimpleId( entity.getUuid(), entity.getType() );

        if (logger.isTraceEnabled()) {
            logger.trace("isCollectionMember(): Checking for edge type {} from {}:{} to {}:{}",
                collectionName, headEntity.getType(), headEntity.getUuid(), entity.getType(), entity.getUuid() );
        }

        GraphManager gm = managerCache.getGraphManager( applicationScope );
        Observable<MarkedEdge> edges = gm.loadEdgeVersions( CpNamingUtils
            .createEdgeFromCollectionName( new SimpleId( headEntity.getUuid(), headEntity.getType() ), collectionName,
                entityId ) );

        return edges.toBlocking().firstOrDefault( null ) != null;
    }


    @Override
    public Set<String> getCollections() throws Exception {

        final Set<String> indexes = new HashSet<String>();

        GraphManager gm = managerCache.getGraphManager( applicationScope );

        Observable<String> str =
            gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( cpHeadEntity.getId(), null, null ) );

        Iterator<String> iter = str.toBlocking().getIterator();
        while ( iter.hasNext() ) {
            String edgeType = iter.next();
            indexes.add( getNameFromEdgeType( edgeType ) );
        }

        return indexes;
    }


    @Override
    public Results getCollection( String collectionName, UUID startResult, int count, Level resultsLevel,
                                  boolean reversed ) throws Exception {

        final String ql;


        if (startResult != null ) {

            // UUID timestamp is a different measure than 'created' field on entities
            Calendar uuidEpoch = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
            uuidEpoch.clear();
            uuidEpoch.set(1582, 9, 15, 0, 0, 0); // 9 = October
            long epochMillis = uuidEpoch.getTime().getTime();

            long time = (startResult.timestamp() / 10000L) + epochMillis;

            if ( !reversed ) {
                ql = "select * where created > " + time;
            } else {
                ql = "select * where created < " + time;
            }

        } else {
            ql = "select *";
        }

        Query query = Query.fromQL( ql );
        if(query == null ){
            throw new RuntimeException("Unable to get data for collection: "+collectionName);
        }
        query.setLimit( count );
        query.setReversed( reversed );

        return searchCollection( collectionName, query );
    }


    @Override
    public Results getCollection( String collectionName, Query query, Level level ) throws Exception {

        return searchCollection( collectionName, query );
    }


    // add to a named collection of the head entity
    @Override
    public Entity addToCollection( String collectionName, EntityRef itemRef ) throws Exception {

        Preconditions.checkNotNull( itemRef, "itemref is null" );
        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
        if ( ( collection != null && collection.getType() != null ) && !collection.getType()
                                                                                  .equals( itemRef.getType() ) ) {
            return null;
        }


        Id entityId = new SimpleId( itemRef.getUuid(), itemRef.getType() );
        org.apache.usergrid.persistence.model.entity.Entity memberEntity = ( ( CpEntityManager ) em ).load( entityId );

        Id memberEntityId = memberEntity.getId();

        // don't fetch entity if we've already got one
        final Entity itemEntity;
        if ( itemRef instanceof Entity ) {
            itemEntity = ( Entity ) itemRef;
        }
        else {
            itemEntity = em.get( itemRef );
        }

        if ( itemEntity == null ) {
            return null;
        }


        if ( memberEntityId == null ) {
            throw new RuntimeException(
                "Unable to load entity uuid=" + itemRef.getUuid() + " type=" + itemRef.getType() );
        }

        if ( logger.isDebugEnabled() ) {
            logger.debug( "Loaded member entity {}:{} from   app {}\n    data {}",
                itemRef.getType(), itemRef.getUuid(), applicationScope, CpEntityMapUtils.toMap( memberEntity ) );
        }


        // create graph edge connection from head entity to member entity
        final Edge edge = createCollectionEdge( cpHeadEntity.getId(), collectionName, memberEntityId );
        final String linkedCollection = collection.getLinkedCollection();

        GraphManager gm = managerCache.getGraphManager( applicationScope );

        gm.writeEdge( edge ).doOnNext( writtenEdge -> {
            if ( logger.isDebugEnabled() ) {
                logger.debug( "Wrote edge {}", writtenEdge );
            }
        } ).filter( writtenEdge -> linkedCollection != null ).flatMap( writtenEdge -> {
            final String pluralType = InflectionUtils.pluralize( cpHeadEntity.getId().getType() );
            final Edge reverseEdge = createCollectionEdge( memberEntityId, pluralType, cpHeadEntity.getId() );

            //reverse
            return gm.writeEdge( reverseEdge ).doOnNext( reverseEdgeWritten -> {

                String entityType = cpHeadEntity.getId().getType();
                if ( !skipIndexingForType( entityType) ) {
                    QueueIndexingStrategy queueIndexingStrategy = getIndexingStrategyForType(entityType);
                    indexService.queueNewEdge(applicationScope, cpHeadEntity.getId(), reverseEdge, queueIndexingStrategy);
                }

            } );
        } ).doOnCompleted( () -> {

            String entityType = memberEntity.getId().getType();
            if ( !skipIndexingForType( entityType ) ) {
                QueueIndexingStrategy queueIndexingStrategy = getIndexingStrategyForType(entityType);
                indexService.queueNewEdge(applicationScope, memberEntityId, edge, queueIndexingStrategy);
            }


            if ( logger.isDebugEnabled() ) {
                logger.debug( "Added entity {}:{} to collection {}",
                    itemRef.getUuid().toString(), itemRef.getType(), collectionName );
            }
        } ).toBlocking().lastOrDefault( null );


        // remove any duplicate edges (keeps the duplicate edge with same timestamp)
        removeDuplicateEdgesAsync(gm, edge);


        if ( logger.isDebugEnabled() ) {
            logger.debug( "Added entity {}:{} to collection {}",
                itemRef.getUuid().toString(), itemRef.getType(), collectionName  );
        }


        return itemEntity;
    }


    @Override
    public Entity createItemInCollection( String collectionName, String itemType, Map<String, Object> properties )
        throws Exception {

        if ( headEntity.getUuid().equals( applicationId ) ) {
            if ( itemType.equals( TYPE_ENTITY ) ) {
                itemType = singularize( collectionName );
            }

            if ( itemType.equals( TYPE_ROLE ) ) {
                Long inactivity = ( Long ) properties.get( PROPERTY_INACTIVITY );
                if ( inactivity == null ) {
                    inactivity = 0L;
                }
                return em.createRole( ( String ) properties.get( PROPERTY_NAME ),
                    ( String ) properties.get( PROPERTY_TITLE ), inactivity );
            }
            return em.create( itemType, properties );
        }

        else if ( headEntity.getType().equals( Group.ENTITY_TYPE ) && ( collectionName.equals( COLLECTION_ROLES ) ) ) {
            UUID groupId = headEntity.getUuid();
            String roleName = ( String ) properties.get( PROPERTY_NAME );
            return em.createGroupRole( groupId, roleName, ( Long ) properties.get( PROPERTY_INACTIVITY ) );
        }

        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
        if ( ( collection != null ) && !collection.getType().equals( itemType ) ) {
            return null;
        }

        properties = getDefaultSchema().cleanUpdatedProperties( itemType, properties, true );

        Entity itemEntity = em.create( itemType, properties );

        if ( itemEntity != null ) {

            addToCollection( collectionName, itemEntity );

            if ( collection != null && collection.getLinkedCollection() != null ) {
                Id itemEntityId = new SimpleId( itemEntity.getUuid(), itemEntity.getType() );
                final Edge edge = createCollectionEdge( cpHeadEntity.getId(), collectionName, itemEntityId );

                GraphManager gm = managerCache.getGraphManager( applicationScope );
                gm.writeEdge( edge );
            }
        }

        return itemEntity;
    }


    @Override
    public void removeFromCollection( String collectionName, EntityRef itemRef ) throws Exception {

        if ( headEntity.getUuid().equals( applicationId ) ) {
            if ( collectionName.equals( COLLECTION_ROLES ) ) {
                // special handling for roles collection of the application
                Entity itemEntity = em.get( itemRef );
                if ( itemEntity != null ) {
                    RoleRef roleRef = SimpleRoleRef.forRoleEntity( itemEntity );
                    em.deleteRole(roleRef.getApplicationRoleName(), Optional.fromNullable(itemEntity) );
                    return;
                }
            }
            // handles normal app collection deletes
            em.delete( itemRef );
            return;
        }

        // headEntity is not an application (used for management entities and entity collections like user devices)

        if ( logger.isDebugEnabled() ) {
            logger.debug( "Loading entity to remove from collection {}:{} from app {}\n",
                itemRef.getType(), itemRef.getUuid(), applicationScope );
        }

        Id entityId = new SimpleId( itemRef.getUuid(), itemRef.getType() );


        // this will remove the edges from app->entity(collection)
        removeItemFromCollection(collectionName, itemRef);

        // special handling for roles collection of a group
        if ( headEntity.getType().equals( Group.ENTITY_TYPE ) ) {

            if ( collectionName.equals( COLLECTION_ROLES ) ) {
                String path = ( String ) ( ( Entity ) itemRef ).getMetadata( "path" );

                if ( path.startsWith( "/roles/" ) ) {

                    Entity itemEntity =
                        em.get( new SimpleEntityRef( entityId.getType(), entityId.getUuid() ) );

                    RoleRef roleRef = SimpleRoleRef.forRoleEntity( itemEntity );
                    em.deleteRole( roleRef.getApplicationRoleName(), Optional.fromNullable(itemEntity) );
                }
            }
        }
    }

    @Override
    public void removeItemFromCollection( String collectionName, EntityRef itemRef ) throws Exception {

        Id entityId = new SimpleId( itemRef.getUuid(), itemRef.getType() );

        // remove edge from collection to item
        GraphManager gm = managerCache.getGraphManager( applicationScope );


        // mark the edge versions and take the first for later delete edge queue event ( load is descending )
        final Edge markedSourceEdge = gm.loadEdgeVersions(
            CpNamingUtils.createEdgeFromCollectionName( cpHeadEntity.getId(), collectionName, entityId ) )
            .flatMap(edge -> gm.markEdge(edge)).toBlocking().firstOrDefault(null);


        Edge markedReversedEdge = null;
        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
        if (collection != null && collection.getLinkedCollection() != null) {
            // delete reverse edges
            final String pluralType = InflectionUtils.pluralize( cpHeadEntity.getId().getType() );
            markedReversedEdge = gm.loadEdgeVersions(
                CpNamingUtils.createEdgeFromCollectionName( entityId, pluralType, cpHeadEntity.getId() ) )
                .flatMap(reverseEdge -> gm.markEdge(reverseEdge)).toBlocking().firstOrDefault(null);
        }


        /**
         * Remove from the index.  This will call gm.deleteEdge which also deletes the reverse edge(s) and de-indexes
         * older versions of the edge(s).
         *
         */
        if( markedSourceEdge != null ) {
            indexService.queueDeleteEdge(applicationScope, markedSourceEdge);
        }
        if( markedReversedEdge != null ){
            indexService.queueDeleteEdge(applicationScope, markedReversedEdge);

        }

    }


    @Override
    public void copyRelationships( String srcRelationName, EntityRef dstEntityRef, String dstRelationName )
        throws Exception {

        headEntity = em.validate( headEntity );
        dstEntityRef = em.validate( dstEntityRef );

        CollectionInfo srcCollection = getDefaultSchema().getCollection( headEntity.getType(), srcRelationName );

        CollectionInfo dstCollection = getDefaultSchema().getCollection( dstEntityRef.getType(), dstRelationName );

        Results results = null;
        do {
            if ( srcCollection != null ) {
                results = em.getCollection( headEntity, srcRelationName, null, 5000, Level.REFS, false );
            }
            else {
                results = em.getTargetEntities( headEntity, srcRelationName, null, Level.REFS );
            }

            if ( ( results != null ) && ( results.size() > 0 ) ) {
                List<EntityRef> refs = results.getRefs();
                for ( EntityRef ref : refs ) {
                    if ( dstCollection != null ) {
                        em.addToCollection( dstEntityRef, dstRelationName, ref );
                    }
                    else {
                        em.createConnection( dstEntityRef, dstRelationName, ref );
                    }
                }
            }
        }
        while ( ( results != null ) && ( results.hasMoreResults() ) );
    }


    @Override
    public Results searchCollection( String collectionName, Query query ) throws Exception {

        if ( query == null ) {
            query = new Query();
            query.setCollection( collectionName );
        }

        headEntity = em.validate( headEntity );

        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );

        if ( collection == null ) {
            throw new RuntimeException(
                "Cannot find collection-info for '" + collectionName + "' of " + headEntity.getType() + ":" + headEntity
                    .getUuid() );
        }


        query.setEntityType( collection.getType() );
        final Query toExecute = adjustQuery( query );
        final Optional<String> queryString = query.isGraphSearch()? Optional.<String>absent(): query.getQl();
        final Id ownerId = headEntity.asId();
        final boolean analyzeOnly = query.getAnalyzeOnly();
        final boolean returnQuery = query.getReturnQuery();


        if(query.getLevel() == Level.IDS ){

            return new IdQueryExecutor( toExecute.getCursor() ) {
                @Override
                protected Observable<ResultsPage<Id>> buildNewResultsPage(
                    final Optional<String> cursor ) {

                    final CollectionSearch search =
                        new CollectionSearch( applicationScope, ownerId, collectionName, collection.getType(), toExecute.getLimit(),
                            queryString, cursor );

                    search.setAnalyzeOnly(analyzeOnly);
                    search.setReturnQuery(returnQuery);

                    return collectionService.searchCollectionIds( search );
                }
            }.next();

        }

        //wire the callback so we can get each page
        return new EntityQueryExecutor( toExecute.getCursor() ) {
            @Override
            protected Observable<ResultsPage<org.apache.usergrid.persistence.model.entity.Entity>> buildNewResultsPage(
                final Optional<String> cursor ) {

                final CollectionSearch search =
                    new CollectionSearch( applicationScope, ownerId, collectionName, collection.getType(), toExecute.getLimit(),
                        queryString, cursor );

                search.setAnalyzeOnly(analyzeOnly);
                search.setReturnQuery(returnQuery);
                IndexConsistency indexConsistency = getIndexConsistencyForType(collectionName);
                search.setKeepStaleEntries(indexConsistency == IndexConsistency.LATEST);

                return collectionService.searchCollection( search );
            }
        }.next();
    }


    @Override
    public Results searchCollectionConsistent( String collectionName, Query query, int expectedResults )
        throws Exception {
        Results results;
        long maxLength = entityManagerFig.pollForRecordsTimeout();
        long sleepTime = entityManagerFig.sleep();
        boolean found;
        long current = System.currentTimeMillis(), length = 0;
        do {
            results = searchCollection( collectionName, query );
            length = System.currentTimeMillis() - current;
            found = expectedResults == results.size();
            if ( found ) {
                break;
            }
            logger.info("Sleeping {} ms during searchCollectionConsistent", sleepTime);
            Thread.sleep( sleepTime );
        }
        while ( !found && length <= maxLength );
        logger.info( "Consistent Search finished in {}, results={}, expected={}...dumping stack",
            length, results.size(), expectedResults );
        return results;
    }


    @Override
    public ConnectionRef createConnection( ConnectionRef connection ) throws Exception {

        return createConnection( connection.getConnectionType(), connection.getTargetRefs() );
    }


    @Override
    public ConnectionRef createConnection( String connectionType, EntityRef connectedEntityRef ) throws Exception {

        headEntity = em.validate( headEntity );
        connectedEntityRef = em.validate( connectedEntityRef );

        ConnectionRefImpl connection = new ConnectionRefImpl( headEntity, connectionType, connectedEntityRef );

        if ( logger.isTraceEnabled() ) {
            logger.trace( "createConnection(): Indexing connection type '{}'\n   from source {}:{}]\n   to target {}:{}\n   app {}",
                connectionType, headEntity.getType(), headEntity.getUuid(), connectedEntityRef.getType(),
                connectedEntityRef.getUuid(), applicationScope );
        }

        final Id entityId = new SimpleId( connectedEntityRef.getUuid(), connectedEntityRef.getType() );
        final org.apache.usergrid.persistence.model.entity.Entity targetEntity =
            ( ( CpEntityManager ) em ).load( entityId );

        // create graph edge connection from head entity to member entity
        final Edge edge = createConnectionEdge( cpHeadEntity.getId(), connectionType, targetEntity.getId() );

        final GraphManager gm = managerCache.getGraphManager( applicationScope );


        //write new edge

        gm.writeEdge(edge).toBlocking().lastOrDefault(null); //throw an exception if this fails


        String entityType = targetEntity.getId().getType();
        if ( !skipIndexingForType( entityType ) ) {
            QueueIndexingStrategy queueIndexingStrategy = getIndexingStrategyForType(entityType);
            indexService.queueNewEdge(applicationScope, targetEntity.getId(), edge, queueIndexingStrategy);
        }

        // remove any duplicate edges (keeps the duplicate edge with same timestamp)
        removeDuplicateEdgesAsync(gm, edge);


        return connection;
    }


    @Override
    public ConnectionRef createConnection( String pairedConnectionType, EntityRef pairedEntity, String connectionType,
                                           EntityRef connectedEntityRef ) throws Exception {

        throw new UnsupportedOperationException( "Paired connections not supported" );
    }


    @Override
    public ConnectionRef createConnection( ConnectedEntityRef... connections ) throws Exception {

        throw new UnsupportedOperationException( "Paired connections not supported" );
    }


    @Override
    public ConnectionRef connectionRef( String connectionType, EntityRef connectedEntityRef ) throws Exception {

        ConnectionRef connection = new ConnectionRefImpl( headEntity, connectionType, connectedEntityRef );

        return connection;
    }


    @Override
    public ConnectionRef connectionRef( String pairedConnectionType, EntityRef pairedEntity, String connectionType,
                                        EntityRef connectedEntityRef ) throws Exception {

        throw new UnsupportedOperationException( "Paired connections not supported" );
    }


    @Override
    public ConnectionRef connectionRef( ConnectedEntityRef... connections ) {

        throw new UnsupportedOperationException( "Paired connections not supported" );
    }


    @Override
    public void deleteConnection( ConnectionRef connectionRef ) throws Exception {

        // First, clean up the dictionary records of the connection
        EntityRef connectingEntityRef = connectionRef.getSourceRefs();  // source
        EntityRef connectedEntityRef = connectionRef.getTargetRefs();  // target

        String connectionType = connectionRef.getTargetRefs().getConnectionType();


        if ( logger.isTraceEnabled() ) {
            logger.trace( "Deleting connection '{}' from source {}:{} \n   to target {}:{}",
                connectionType, connectingEntityRef.getType(), connectingEntityRef.getUuid(),
                connectedEntityRef.getType(), connectedEntityRef.getUuid() );
        }

        Id entityId = new SimpleId( connectedEntityRef.getUuid(), connectedEntityRef.getType() );
        org.apache.usergrid.persistence.model.entity.Entity targetEntity = ( ( CpEntityManager ) em ).load( entityId );

        GraphManager gm = managerCache.getGraphManager( applicationScope );

        final Id sourceId = new SimpleId( connectingEntityRef.getUuid(), connectingEntityRef.getType() );

        final SearchByEdge search = createConnectionSearchByEdge( sourceId, connectionType, targetEntity.getId() );

        //delete all the edges and queue their processing
        gm.loadEdgeVersions( search ).flatMap( returnedEdge -> gm.markEdge( returnedEdge ) )
          .doOnNext( returnedEdge -> {

              if ( !skipIndexingForType( returnedEdge.getSourceNode().getType() ) || !skipIndexingForType( returnedEdge.getTargetNode().getType() ) ) {

                  indexService.queueDeleteEdge(applicationScope, returnedEdge);
              }

          }).toBlocking()
          .lastOrDefault( null );
    }


    @Override
    public Set<String> getConnectionTypes( UUID connectedEntityId ) throws Exception {
        throw new UnsupportedOperationException( "Cannot specify entity by UUID alone." );
    }


    @Override
    public Set<String> getConnectionTypes() throws Exception {
        return getConnectionTypes( false );
    }


    @Override
    public Set<String> getConnectionTypes( boolean filterConnection ) throws Exception {

        final GraphManager gm = managerCache.getGraphManager( applicationScope );

        Observable<String> edges =
            gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( cpHeadEntity.getId(), null, null ) );

        return edges.collect( () -> new HashSet<String>(), ( edgeSet, edge ) -> {
            edgeSet.add( CpNamingUtils.getNameFromEdgeType( edge ) );

        } ).toBlocking().last();

    }


    @Override
    public Results getTargetEntities( String connectionType, String connectedEntityType, Level level )
        throws Exception {

        //until this is refactored properly, we will delegate to a search by query
        Results raw = null;

        Preconditions.checkNotNull( connectionType, "connectionType cannot be null" );

        Query query = new Query();
        query.setConnectionType( connectionType );
        query.setEntityType( connectedEntityType );
        query.setResultsLevel( level );

        return searchTargetEntities( query );
    }


    @Override
    public Results getSourceEntities( String connType, String fromEntityType, Level resultsLevel ) throws Exception {

        return getSourceEntities( connType, fromEntityType, resultsLevel, -1 );
    }


    @Override
    public Results getSourceEntities( String connType, String fromEntityType, Level level, int count )
        throws Exception {

        // looking for edges to the head entity
        String edgeType = CpNamingUtils.getEdgeTypeFromConnectionType( connType );

        Map<EntityRef, Set<String>> containers = getContainers( count, edgeType, fromEntityType );

        if ( Level.REFS.equals( level ) ) {
            List<EntityRef> refList = new ArrayList<EntityRef>( containers.keySet() );
            return Results.fromRefList( refList );
        }

        if ( Level.IDS.equals( level ) ) {
            // TODO: someday this should return a list of Core Persistence Ids
            List<UUID> idList = new ArrayList<UUID>();
            for ( EntityRef ref : containers.keySet() ) {
                idList.add( ref.getUuid() );
            }
            return Results.fromIdList( idList );
        }

        List<Entity> entities = new ArrayList<Entity>();
        for ( EntityRef ref : containers.keySet() ) {
            Entity entity = em.get( ref );

            if (logger.isTraceEnabled()) {
                logger.trace("   Found connecting entity: " + entity.getProperties());
            }

            entities.add( entity );
        }
        return Results.fromEntities( entities );
    }


    @Override
    public Results searchTargetEntities( Query query ) throws Exception {

        Preconditions.checkNotNull( query, "query cannot be null" );

        final String connection = query.getConnectionType();

        Preconditions.checkNotNull( connection, "connection must be specified" );

        headEntity = em.validate( headEntity );

        final boolean analyzeOnly = query.getAnalyzeOnly();
        final boolean returnQuery = query.getReturnQuery();


        final Query toExecute = adjustQuery( query );

        final Optional<String> entityType = Optional.fromNullable( query.getEntityType() );
        //set startid -- graph | es query filter -- load entities filter (verifies exists) --> results page collector
        // -> 1.0 results

        //  startid -- graph edge load -- entity load (verify) from ids -> results page collector
        // startid -- eq query candiddate -- entity load (verify) from canddiates -> results page collector

        //startid -- graph edge load -- entity id verify --> filter to connection ref --> connection ref collector
        //startid -- eq query candiddate -- candidate id verify --> filter to connection ref --> connection ref
        // collector

        final Id sourceId = headEntity.asId();

        final Optional<String> queryString = query.isGraphSearch()? Optional.<String>absent(): query.getQl();
        final boolean isConnecting = query.isConnecting();

        if ( query.getResultsLevel() == Level.REFS || query.getResultsLevel() == Level.IDS ) {


            return new ConnectionRefQueryExecutor( toExecute.getCursor() ) {
                @Override
                protected Observable<ResultsPage<ConnectionRef>> buildNewResultsPage( final Optional<String> cursor ) {


                //we need the callback so as we get a new cursor, we execute a new search and re-initialize our builders

                    final ConnectionSearch search =
                        new ConnectionSearch( applicationScope, sourceId, entityType, connection, toExecute.getLimit(),
                            queryString, cursor, isConnecting );
                    search.setAnalyzeOnly(analyzeOnly);
                    search.setReturnQuery(returnQuery);
                    return connectionService.searchConnectionAsRefs( search );
                }
            }.next();
        }


        return new EntityQueryExecutor( toExecute.getCursor() ) {
            @Override
            protected Observable<ResultsPage<org.apache.usergrid.persistence.model.entity.Entity>> buildNewResultsPage(
                final Optional<String> cursor ) {

                //we need the callback so as we get a new cursor, we execute a new search and re-initialize our builders
                final ConnectionSearch search =
                    new ConnectionSearch( applicationScope, sourceId, entityType, connection, toExecute.getLimit(),
                        queryString, cursor, isConnecting );
                search.setAnalyzeOnly(analyzeOnly);
                search.setReturnQuery(returnQuery);
                return connectionService.searchConnection( search );
            }
        }.next();
    }


    private Query adjustQuery( Query query ) {

        // handle the select by identifier case
        if ( query.getQl().isPresent() ) {
            return query;
        }

        // a name alias or email alias was specified
        if ( query.containsSingleNameOrEmailIdentifier() ) {

            Identifier ident = query.getSingleIdentifier();

            // an email was specified.  An edge case that only applies to users.
            // This is fulgy to put here, but required.
            if ( query.getEntityType().equals( User.ENTITY_TYPE ) && ident.isEmail() ) {

                final String newQuery = "select * where email='" + query.getSingleNameOrEmailIdentifier() + "'";

                query.setQl( newQuery );
            }
            // groups have a special unique identifier
            else if ( query.getEntityType().equals( Group.ENTITY_TYPE ) ){

                final String newQuery = "select * where path='" + query.getSingleNameOrEmailIdentifier() + "'";

                query.setQl( newQuery );
            }

            // use the ident with the default alias. could be an email
            else {

                final String newQuery = "select * where name='" + query.getSingleNameOrEmailIdentifier() + "'";
                query.setQl( newQuery );
            }
        }
        else if ( query.containsSingleUuidIdentifier() ) {

            //TODO, this shouldn't even come from ES, it should look up the entity directly
            final String newQuery = "select * where uuid=" + query.getSingleUuidIdentifier() + "";
            query.setQl( newQuery );
        }


        //TODO T.N. not sure if we still need this.  If we do then we ned to modify our query interface.

        //        final String ql = query.getQl();
        //
        //        if ( query.isReversed()  && ( StringUtils.isEmpty( ql ) || !(ql.contains( "order by " )) )) {
        //
        //
        //            final String sortQueryString =
        //
        //            SortPredicate
        //            Query.SortPredicate desc =
        //                new Query.SortPredicate( PROPERTY_CREATED, Query.SortDirection.DESCENDING );
        //
        //            try {
        //                query.addSort( desc );
        //            }
        //            catch ( Exception e ) {
        //                logger.warn( "Attempted to reverse sort order already set", PROPERTY_CREATED );
        //            }
        //        }


        return query;
    }


    @Override
    public Set<String> getConnectionIndexes( String connectionType ) throws Exception {
        throw new UnsupportedOperationException( "Not supported yet." );
    }


    /** side effect: converts headEntity into an Entity if it is an EntityRef! */
    private Entity getHeadEntity() throws Exception {
        Entity entity = null;
        if ( headEntity instanceof Entity ) {
            entity = ( Entity ) headEntity;
        }
        else {
            entity = em.get( headEntity );
            headEntity = entity;
        }
        return entity;
    }

    private void removeDuplicateEdgesAsync(GraphManager gm, Edge edge){

        //now read all older versions of an edge, and remove them.  Finally calling delete
        final SearchByEdge searchByEdge =
            new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE,
                SearchByEdgeType.Order.DESCENDING, Optional.absent() );

        //load our versions, only retain the most recent one
        gm.loadEdgeVersions(searchByEdge).skip(1).flatMap(edgeToDelete -> {
            if (logger.isDebugEnabled()) {
                logger.debug("Duplicate edge. Marking edge {} for deletion", edgeToDelete);
            }
            return gm.markEdge(edgeToDelete );
        }).lastOrDefault(null).doOnNext(lastEdge -> {
            //no op if we hit our default
            if (lastEdge == null) {
                return;
            }
            //don't queue delete b/c that de-indexes, we need to delete the edges only since we have a version still existing to index.
            gm.deleteEdge(lastEdge).toBlocking().lastOrDefault(null); // this should throw an exception
        }).toBlocking().lastOrDefault(null);//this should throw an exception

    }

    private IndexConsistency getIndexConsistencyForType(String type ) {
        return CpCollectionUtils.getIndexConsistencyForType(collectionSettingsFactory, applicationId, type);
    }

    private QueueIndexingStrategy getIndexingStrategyForType(String type ) {
        return CpCollectionUtils.getIndexingStrategyForType(collectionSettingsFactory, applicationId, type);

    }

    private boolean skipIndexingForType( String type ) {
        return CpCollectionUtils.skipIndexingForType(collectionSettingsFactory, applicationId, type);
    }

    /**
     * Get the map manager for uuid mapping
     */
    private MapManager getMapManagerForTypes() {
        Id mapOwner = new SimpleId( applicationId, TYPE_APPLICATION );

        final MapScope ms = CpNamingUtils.getEntityTypeMapScope(mapOwner);

        MapManager mm = managerCache.getMapManager( ms );

        return mm;
    }

}