/*
 * The MIT License (MIT)
 *
 * Copyright (c) Open Application Platform Authors
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package oap.storage;

import com.fasterxml.jackson.core.type.TypeReference;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.ReplaceOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import oap.util.Lists;
import oap.util.Stream;
import org.apache.commons.lang3.mutable.MutableInt;
import org.bson.codecs.configuration.CodecRegistries;
import org.bson.codecs.configuration.CodecRegistry;
import org.bson.types.ObjectId;
import org.joda.time.DateTimeUtils;

import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;

import static com.mongodb.client.model.Filters.eq;

/**
 * Created by igor.petrenko on 15.09.2017.
 */
@Slf4j
public class MongoStorage<T> extends MemoryStorage<T> implements Runnable {
    public static final UpdateOptions UPDATE_OPTIONS_UPSERT = new UpdateOptions().upsert( true );
    public final MongoCollection<Item<T>> collection;
    public final MongoDatabase database;
    public int bulkSize = 1000;
    private long lastFsync = -1;

    @SuppressWarnings( "unchecked" )
    public MongoStorage( oap.storage.MongoClient mongoClient, String database, String table,
                         LockStrategy lockStrategy, Class<T> clazz ) {
        super( IdentifierBuilder
            .annotation()
            .suggestion( ar -> ObjectId.get().toString() )
            .size( 24 )
            .idOptions()
            .build(), lockStrategy );
        this.database = mongoClient.getDatabase( database );


        final Object o = new TypeReference<Item<T>>() {};
        final CodecRegistry codecRegistry = CodecRegistries.fromRegistries(
            CodecRegistries.fromCodecs( new JsonCodec<>( ( TypeReference<Item> ) o,
                Item.class, clazz, ( m ) -> identifier.get( ( T ) m.object ) ) ),
            this.database.getCodecRegistry()
        );

        final Object metadataMongoCollection = this.database
            .getCollection( table, Item.class )
            .withCodecRegistry( codecRegistry );
        this.collection = ( MongoCollection<Item<T>> ) metadataMongoCollection;

        load();
    }

    private void load() {
        lastFsync = DateTimeUtils.currentTimeMillis();

        final Consumer<Item<T>> cons = metadata -> {
            val id = identifier.get( metadata.object );
            data.put( id, metadata );
        };

        log.info( "[{}] total {}", getStorageId(), collection.count() );

        collection.find().forEach( cons );
    }

    @Override
    public void close() {
        fsync();

        super.close();
    }

    @Override
    public void fsync() {
        super.fsync();

        val count = new MutableInt();

        Stream.of( data.values()
            .stream()
            .filter( m -> m.modified >= lastFsync ) )
            .grouped( bulkSize )
            .forEach( list -> {
                count.add( list.size() );

                final List<? extends WriteModel<Item<T>>> bulk = Lists.map( list,
                    metadata -> {
                        val id = identifier.get( metadata.object );
                        return new ReplaceOneModel<>( eq( "_id", new ObjectId( id ) ), metadata, UPDATE_OPTIONS_UPSERT );
                    } );
                collection.bulkWrite( bulk );

            } );

        log.info( "[{}] fsync total: {}, modified: {}", getStorageId(), size(), count.intValue() );
        lastFsync = System.currentTimeMillis();
    }

    private String getStorageId() {
        return database.getName() + "." + collection.getNamespace();
    }

    @Override
    public void run() {
        fsync();
    }

    @Override
    public Optional<T> delete( String id ) {
        collection.deleteOne( eq( "_id", new ObjectId( id ) ) );

        return super.delete( id );
    }
}

@@ -3674,36 +3674,47 @@
  total %7B%7D%22, 
-getStorageId
+collection.getNamespace
 (), collecti
@@ -4638,28 +4638,39 @@
 d: %7B%7D%22, 
-getStorageId
+collection.getNamespace
 (), size
@@ -4752,120 +4752,8 @@
  %7D%0A%0A
-    private String getStorageId() %7B%0A        return database.getName() + %22.%22 + collection.getNamespace();%0A    %7D%0A%0A
     

update logging