/*
 * Copyright 2019, TeamDev. 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 io.spine.system.server;

import io.spine.core.Command;
import io.spine.core.CommandId;
import io.spine.core.Event;
import io.spine.core.EventId;
import io.spine.server.entity.RecentHistory;
import io.spine.system.server.event.CommandDispatchedToHandler;
import io.spine.system.server.event.EventDispatchedToReactor;
import io.spine.system.server.event.EventDispatchedToSubscriber;
import io.spine.type.TypeUrl;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * Performs a lookup over a given recent history to tell whether or not a given message has already
 * been dispatched to the given entity.
 */
final class DuplicateLookup {

    private static final TypeUrl COMMAND_DISPATCHED_EVENT_TYPE =
            TypeUrl.from(CommandDispatchedToHandler.getDescriptor());

    private static final TypeUrl EVENT_DISPATCHED_TO_REACTOR_EVENT_TYPE =
            TypeUrl.from(EventDispatchedToReactor.getDescriptor());

    private static final TypeUrl EVENT_DISPATCHED_TO_SUBSCRIBER_EVENT_TYPE =
            TypeUrl.from(EventDispatchedToSubscriber.getDescriptor());

    private final RecentHistory history;

    private DuplicateLookup(RecentHistory history) {
        this.history = history;
    }

    /**
     * Creates a new lookup through the given {@link RecentHistory}.
     *
     * @param history the history of entity to search for duplicates in
     * @return new instance of {@code DuplicateLookup}
     */
    static DuplicateLookup through(RecentHistory history) {
        checkNotNull(history);
        return new DuplicateLookup(history);
    }

    /**
     * Tells if the given event has already been dispatched to the given entity.
     *
     * @param candidate
     *         the event to check
     * @return {@code true} if the given event is a duplicate, {@code false} otherwise
     */
    boolean isDuplicate(Event candidate) {
        return isDuplicateOfReact(candidate) || isDuplicateOfSubscribe(candidate);
    }

    private boolean isDuplicateOfReact(Event candidate) {
        EventId candidateId = candidate.getId();
        boolean duplicate = history.stream()
                                   .filter(DuplicateLookup::isEventDispatchedToReactor)
                                   .map(Event::enclosedMessage)
                                   .map(EventDispatchedToReactor.class::cast)
                                   .map(EventDispatchedToReactor::getPayload)
                                   .map(Event::id)
                                   .anyMatch(candidateId::equals);
        return duplicate;
    }

    private boolean isDuplicateOfSubscribe(Event candidate) {
        EventId candidateId = candidate.getId();
        boolean duplicate = history.stream()
                                   .filter(DuplicateLookup::isEventDispatchedToSubscriber)
                                   .map(Event::enclosedMessage)
                                   .map(EventDispatchedToSubscriber.class::cast)
                                   .map(EventDispatchedToSubscriber::getPayload)
                                   .map(Event::id)
                                   .anyMatch(candidateId::equals);
        return duplicate;
    }

    /**
     * Tells if the given command has already been dispatched to the given entity.
     *
     * @param candidate
     *         the command to check
     * @return {@code true} if the given command is a duplicate, {@code false} otherwise
     */
    boolean isDuplicate(Command candidate) {
        CommandId candidateId = candidate.getId();
        boolean duplicate = history.stream()
                                   .filter(DuplicateLookup::isCommandDispatchedToHandler)
                                   .map(Event::enclosedMessage)
                                   .map(CommandDispatchedToHandler.class::cast)
                                   .map(CommandDispatchedToHandler::getPayload)
                                   .map(Command::id)
                                   .anyMatch(candidateId::equals);
        return duplicate;
    }

    private static boolean isEventDispatchedToSubscriber(Event candidate) {
        return instanceOf(candidate, EVENT_DISPATCHED_TO_SUBSCRIBER_EVENT_TYPE);
    }

    private static boolean isEventDispatchedToReactor(Event candidate) {
        return instanceOf(candidate, EVENT_DISPATCHED_TO_REACTOR_EVENT_TYPE);
    }

    private static boolean isCommandDispatchedToHandler(Event candidate) {
        return instanceOf(candidate, COMMAND_DISPATCHED_EVENT_TYPE);
    }

    private static boolean instanceOf(Event event, TypeUrl type) {
        TypeUrl actualType = event.typeUrl();
        return type.equals(actualType);
    }
}

@@ -3103,32 +3103,48 @@
 lean duplicate =
+%0A               
  history.stream(
@@ -3141,36 +3141,24 @@
 ry.stream()%0A
-            
             
@@ -3240,36 +3240,24 @@
             
-     
-       
 .map(Event::
@@ -3292,36 +3292,24 @@
             
-          
-  
 .map(EventDi
@@ -3335,36 +3335,24 @@
 lass::cast)%0A
-            
             
@@ -3424,36 +3424,24 @@
             
-  
-          
 .map(Event::
@@ -3440,36 +3440,24 @@
 (Event::id)%0A
-            
             
@@ -3662,32 +3662,48 @@
 lean duplicate =
+%0A               
  history.stream(
@@ -3700,36 +3700,24 @@
 ry.stream()%0A
-            
             
@@ -3802,36 +3802,24 @@
             
-            
 .map(Event::
@@ -3854,36 +3854,24 @@
             
-    
-        
 .map(EventDi
@@ -3900,36 +3900,24 @@
 lass::cast)%0A
-            
             
@@ -3992,36 +3992,24 @@
             
-  
-          
 .map(Event::
@@ -4008,36 +4008,24 @@
 (Event::id)%0A
-            
             
@@ -4478,16 +4478,32 @@
 licate =
+%0A               
  history
@@ -4531,36 +4531,24 @@
             
-      
-      
 .filter(Dupl
@@ -4609,36 +4609,24 @@
             
-           
- 
 .map(Event::
@@ -4638,36 +4638,24 @@
 sedMessage)%0A
-            
             
@@ -4729,36 +4729,24 @@
             
-   
-         
 .map(Command
@@ -4797,36 +4797,24 @@
             
-            
 .map(Command
@@ -4815,36 +4815,24 @@
 ommand::id)%0A
-            
             

Improve code layout