/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you 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.polygene.index.elasticsearch;

import java.util.HashMap;
import java.util.Map;
import javax.json.JsonArrayBuilder;
import javax.json.JsonObject;
import javax.json.JsonObjectBuilder;
import javax.json.JsonValue;
import org.apache.polygene.api.entity.EntityDescriptor;
import org.apache.polygene.api.entity.EntityReference;
import org.apache.polygene.api.identity.Identity;
import org.apache.polygene.api.injection.scope.Service;
import org.apache.polygene.api.injection.scope.Structure;
import org.apache.polygene.api.injection.scope.This;
import org.apache.polygene.api.mixin.Mixins;
import org.apache.polygene.api.serialization.Serializer;
import org.apache.polygene.api.structure.ModuleDescriptor;
import org.apache.polygene.api.time.SystemTime;
import org.apache.polygene.api.usecase.UsecaseBuilder;
import org.apache.polygene.api.util.Classes;
import org.apache.polygene.serialization.javaxjson.JavaxJsonFactories;
import org.apache.polygene.spi.entity.EntityState;
import org.apache.polygene.spi.entity.EntityStatus;
import org.apache.polygene.spi.entity.ManyAssociationState;
import org.apache.polygene.spi.entity.NamedAssociationState;
import org.apache.polygene.spi.entitystore.EntityStore;
import org.apache.polygene.spi.entitystore.EntityStoreUnitOfWork;
import org.apache.polygene.spi.entitystore.StateChangeListener;
import org.apache.polygene.spi.serialization.JsonSerializer;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Listen to Entity state changes and index them in ElasticSearch.
 *
 * QUID Use two indices, one for strict queries, one for full text and fuzzy search?
 */
@Mixins( ElasticSearchIndexer.Mixin.class )
public interface ElasticSearchIndexer extends StateChangeListener
{
    class Mixin
        implements StateChangeListener
    {
        private static final Logger LOGGER = LoggerFactory.getLogger( ElasticSearchIndexer.class );

        @Structure
        private ModuleDescriptor module;

        @Service
        private EntityStore entityStore;

        @Service
        private JsonSerializer jsonSerializer;

        @Service
        private JavaxJsonFactories jsonFactories;

        @This
        private ElasticSearchSupport support;

        public void emptyIndex()
        {
            support.client().admin().indices().prepareDelete( support.index() ).execute().actionGet();
        }

        @Override
        public void notifyChanges( Iterable<EntityState> changedStates )
        {
            // All updated or new states
            Map<String, EntityState> newStates = new HashMap<>();
            for( EntityState eState : changedStates )
            {
                if( eState.status() == EntityStatus.UPDATED || eState.status() == EntityStatus.NEW )
                {
                    newStates.put( eState.entityReference().identity().toString(), eState );
                }
            }

            EntityStoreUnitOfWork uow = entityStore.newUnitOfWork(
                module,
                UsecaseBuilder.newUsecase( "Load associations for indexing" ),
                SystemTime.now()
            );

            // Bulk index request builder
            BulkRequestBuilder bulkBuilder = support.client().prepareBulk();

            // Handle changed entity states
            for( EntityState changedState : changedStates )
            {
                if( changedState.entityDescriptor().queryable() )
                {
                    switch( changedState.status() )
                    {
                        case REMOVED:
                            LOGGER.trace( "Removing Entity State from Index: {}", changedState );
                            remove( bulkBuilder, changedState.entityReference().identity().toString() );
                            break;
                        case UPDATED:
                            LOGGER.trace( "Updating Entity State in Index: {}", changedState );
                            remove( bulkBuilder, changedState.entityReference().identity().toString() );
                            String updatedJson = toJSON( changedState, newStates, uow ).toString();
                            LOGGER.trace( "Will index: {}", updatedJson );
                            index( bulkBuilder, changedState.entityReference().identity().toString(), updatedJson );
                            break;
                        case NEW:
                            LOGGER.trace( "Creating Entity State in Index: {}", changedState );
                            String newJson = toJSON( changedState, newStates, uow ).toString();
                            LOGGER.trace( "Will index: {}", newJson );
                            index( bulkBuilder, changedState.entityReference().identity().toString(), newJson );
                            break;
                        case LOADED:
                        default:
                            // Ignored
                            break;
                    }
                }
            }

            uow.discard();

            if( bulkBuilder.numberOfActions() > 0 )
            {

                // Execute bulk actions
                BulkResponse bulkResponse = bulkBuilder.execute().actionGet();

                // Handle errors
                if( bulkResponse.hasFailures() )
                {
                    throw new ElasticSearchIndexingException( bulkResponse.buildFailureMessage() );
                }

                LOGGER.debug( "Indexing changed Entity states took {}ms", bulkResponse.getTookInMillis() );

                // Refresh index
                support.client().admin().indices().prepareRefresh( support.index() ).execute().actionGet();
            }
        }

        private void remove( BulkRequestBuilder bulkBuilder, String identity )
        {
            bulkBuilder.add( support.client().
                prepareDelete( support.index(), support.entitiesType(), identity ) );
        }

        private void index( BulkRequestBuilder bulkBuilder, String identity, String json )
        {
            bulkBuilder.add( support.client().
                prepareIndex( support.index(), support.entitiesType(), identity ).
                    setSource( json, XContentType.JSON ) );
        }

        /**
         * <pre>
         * {
         *  "_identity": "ENTITY-IDENTITY",
         *  "_types": [ "All", "Entity", "types" ],
         *  "property.name": property.value,
         *  "association.name": { "reference": "ASSOCIATED-IDENTITY" }
         *  "manyassociation.name": [ { "reference": "ASSOCIATED" }, { "reference": "IDENTITIES" } ]
         *  "namedassociation.name": [ { "_named": "NAMED", "reference": "IDENTITY" } }
         * }
         * </pre>
         */
        private JsonObject toJSON( EntityState state, Map<String, EntityState> newStates, EntityStoreUnitOfWork uow )
        {
            JsonObjectBuilder builder = jsonFactories.builderFactory().createObjectBuilder();

            builder.add( "_identity", state.entityReference().identity().toString() );

            JsonArrayBuilder typesBuilder = jsonFactories.builderFactory().createArrayBuilder();
            state.entityDescriptor().mixinTypes().map( Classes.toClassName() ).forEach( typesBuilder::add );
            builder.add( "_types", typesBuilder.build() );

            EntityDescriptor entityType = state.entityDescriptor();

            // Properties
            entityType.state().properties().forEach(
                propDesc ->
                {
                    if( propDesc.queryable() )
                    {
                        String key = propDesc.qualifiedName().name();
                        Object value = state.propertyValueOf( propDesc.qualifiedName() );
                        JsonValue jsonValue = jsonSerializer.toJson( Serializer.Options.NO_TYPE_INFO, value );
                        builder.add( key, jsonValue );
                    }
                } );

            // Associations
            entityType.state().associations().forEach(
                assocDesc ->
                {
                    if( assocDesc.queryable() )
                    {
                        String key = assocDesc.qualifiedName().name();
                        EntityReference associated = state.associationValueOf( assocDesc.qualifiedName() );
                        if( associated == null )
                        {
                            builder.add( key, JsonValue.NULL );
                        }
                        else
                        {
                            if( assocDesc.isAggregated() || support.indexNonAggregatedAssociations() )
                            {
                                if( newStates.containsKey( associated.identity().toString() ) )
                                {
                                    builder.add( key, toJSON( newStates.get( associated.identity().toString() ),
                                                              newStates, uow ) );
                                }
                                else
                                {
                                    EntityReference reference = EntityReference.create( associated.identity() );
                                    EntityState assocState = uow.entityStateOf( entityType.module(), reference );
                                    builder.add( key, toJSON( assocState, newStates, uow ) );
                                }
                            }
                            else
                            {
                                builder.add( key, jsonFactories.builderFactory().createObjectBuilder()
                                                               .add( "reference", associated.identity().toString() ) );
                            }
                        }
                    }
                } );

            // ManyAssociations
            entityType.state().manyAssociations().forEach(
                manyAssocDesc ->
                {
                    if( manyAssocDesc.queryable() )
                    {
                        String key = manyAssocDesc.qualifiedName().name();
                        JsonArrayBuilder assBuilder = jsonFactories.builderFactory().createArrayBuilder();
                        ManyAssociationState assocs = state.manyAssociationValueOf( manyAssocDesc.qualifiedName() );
                        for( EntityReference associated : assocs )
                        {
                            if( manyAssocDesc.isAggregated() || support.indexNonAggregatedAssociations() )
                            {
                                if( newStates.containsKey( associated.identity().toString() ) )
                                {
                                    assBuilder.add( toJSON( newStates.get( associated.identity().toString() ),
                                                            newStates, uow ) );
                                }
                                else
                                {
                                    EntityReference reference = EntityReference.create( associated.identity() );
                                    EntityState assocState = uow.entityStateOf( entityType.module(), reference );
                                    assBuilder.add( toJSON( assocState, newStates, uow ) );
                                }
                            }
                            else
                            {
                                assBuilder.add( jsonFactories.builderFactory().createObjectBuilder()
                                                             .add( "reference",
                                                                   associated.identity().toString() ) );
                            }
                        }
                        builder.add( key, assBuilder.build() );
                    }
                } );

            // NamedAssociations
            entityType.state().namedAssociations().forEach(
                namedAssocDesc ->
                {
                    if( namedAssocDesc.queryable() )
                    {
                        String key = namedAssocDesc.qualifiedName().name();
                        JsonArrayBuilder assBuilder = jsonFactories.builderFactory().createArrayBuilder();
                        NamedAssociationState assocs = state.namedAssociationValueOf(
                            namedAssocDesc.qualifiedName() );
                        for( String name : assocs )
                        {
                            Identity identity = assocs.get( name ).identity();
                            if( namedAssocDesc.isAggregated() || support.indexNonAggregatedAssociations() )
                            {
                                String identityString = identity.toString();
                                if( newStates.containsKey( identityString ) )
                                {
                                    assBuilder.add(
                                        jsonFactories.cloneBuilder( toJSON( newStates.get( identityString ),
                                                                            newStates, uow ) )
                                                     .add( "_named", name )
                                                     .build() );
                                }
                                else
                                {
                                    EntityReference reference = EntityReference.create( identity );
                                    EntityState assocState = uow.entityStateOf( entityType.module(), reference );
                                    assBuilder.add(
                                        jsonFactories.cloneBuilder( toJSON( assocState, newStates, uow ) )
                                                     .add( "_named", name ).build() );
                                }
                            }
                            else
                            {
                                assBuilder.add( jsonFactories.builderFactory().createObjectBuilder()
                                                             .add( "_named", name )
                                                             .add( "reference", identity.toString() )
                                                             .build() );
                            }
                        }
                        builder.add( key, assBuilder.build() );
                    }
                } );
            return builder.build();
        }
    }
}
