/*
 * Copyright 2017, TeamDev Ltd. All rights reserved.
 *
 * Redistribution and use in source and/or binary forms, with or without
 * modification, must retain the above copyright notice and the following
 * disclaimer.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package org.spine3.server.event;

import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Iterators;
import com.google.protobuf.Any;
import com.google.protobuf.FieldMask;
import com.google.protobuf.Timestamp;
import org.spine3.base.Event;
import org.spine3.base.EventId;
import org.spine3.client.EntityFilters;
import org.spine3.protobuf.AnyPacker;
import org.spine3.server.entity.DefaultRecordBasedRepository;

import javax.annotation.Nullable;
import java.util.Iterator;

import static com.google.common.base.Preconditions.checkNotNull;
import static org.spine3.server.event.EventEntity.comparator;

/**
 * A storage used by {@link EventStore} for keeping event data.
 *
 * <p>This class allows to hide implementation details of storing events.
 * {@link EventStore} serves as a facade, hiding the fact that the {@code EventStorage}
 * is a {@code Repository}.
 *
 * @author Alexander Yevsyukov
 * @author Dmytro Dashenkov
 */
class EventStorage extends DefaultRecordBasedRepository<EventId, EventEntity, Event> {

    private static final Function<EventEntity, Event> GET_EVENT =
            new Function<EventEntity, Event>() {
                @Nullable
                @Override
                public Event apply(@Nullable EventEntity input) {
                    if (input == null) {
                        return null;
                    }
                    return input.getState();
                }
            };

    Iterator<Event> iterator(EventStreamQuery query) {
        checkNotNull(query);

        final EntityFilters filters = toEntityFilters(query);
        final Iterable<EventEntity> entities = find(filters, FieldMask.getDefaultInstance());

        // Optimize the filtering by excluding the `after` timestamp
        final EventStreamQuery queryRegardlessTime = query.toBuilder()
                                                          .clearAfter()
                                                          .build();
        final Predicate<EventEntity> detailedLookupFilter = createEntityFilter(queryRegardlessTime);

        final Iterator<EventEntity> entityIterator = FluentIterable.from(entities)
                                                                   .filter(detailedLookupFilter)
                                                                   .toSortedList(comparator())
                                                                   .iterator();
        final Iterator<Event> result = Iterators.transform(entityIterator, getEventFunc());
        return result;
    }

    void store(Event event) {
        final EventEntity entity = new EventEntity(event);
        store(entity);
    }

    static Function<EventEntity, Event> getEventFunc() {
        return GET_EVENT;
    }

    private static Predicate<EventEntity> createEntityFilter(EventStreamQuery query) {
        return new EventEntityMatchesStreamQuery(query);
    }

    /**
     * Creates an instance of {@link EntityFilters} from the given {@link EventStreamQuery}.
     *
     * <p>The resulting filters only contain the filtering by the {@code after} field of the source
     * query compared to the {@link EventEntity#getTime()} column.
     *
     * @param query the source {@link EventStreamQuery} to get the info from
     * @return new instance of {@link EntityFilters} filtering the events by their timestamp
     */
    private static EntityFilters toEntityFilters(EventStreamQuery query) {
        final Timestamp timestamp = query.getAfter();
        final Any wrappedTimestamp = AnyPacker.pack(timestamp);
        final EntityFilters entityFilters =
                EntityFilters.newBuilder()
                             .putColumnFilter(EventEntity.TIME_COLUMN, wrappedTimestamp)
                             .build();
        return entityFilters;
    }

    private static class EventEntityMatchesStreamQuery implements Predicate<EventEntity> {

        private final Predicate<Event> filter;

        private EventEntityMatchesStreamQuery(EventStreamQuery query) {
            this.filter = new MatchesStreamQuery(query);
        }

        @Override
        public boolean apply(@Nullable EventEntity input) {
            if (input == null) {
                return false;
            }
            final Event event = input.getState();
            final boolean result = filter.apply(event);
            return result;
        }
    }
}

@@ -1222,40 +1222,8 @@
 rs;%0A
-import com.google.protobuf.Any;%0A
 impo
@@ -1418,24 +1418,28 @@
 ne3.
-protobuf.AnyPack
+client.TimestampFilt
 er;%0A
@@ -4532,48 +4532,104 @@
 nal 
-Timestamp timestamp = 
+EntityFilters.Builder builder = EntityFilters.newBuilder();%0A        if (
 query.
-get
+has
 After()
-;%0A
+) %7B%0A    
     
@@ -4642,99 +4642,97 @@
 nal 
-Any wrappedTimestamp = AnyPacker.pack(timestamp);%0A        final EntityFilters entity
+Timestamp timestamp = query.getAfter();%0A            final TimestampFilter timestamp
 Filter
-s
  =%0A 
@@ -4746,29 +4746,35 @@
         
-Entity
+    Timestamp
 Filter
-s
 .newBuil
@@ -4812,24 +4812,29 @@
     
-.pu
+      .se
 tColumn
-Filter
+sName
 (Eve
@@ -4857,26 +4857,8 @@
 LUMN
-, wrappedTimestamp
 )%0A  
@@ -4876,32 +4876,216 @@
                 
+      .setValue(timestamp)%0A                                   .build();%0A            builder.setTimeAfter(timestampFilter);%0A        %7D%0A        final EntityFilters entityFilters = builder
 .build();%0A      

Build the entity filters in the EventStorage properly.