/*
 * 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.flink.cep.nfa.compiler;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.State;
import org.apache.flink.cep.nfa.StateTransition;
import org.apache.flink.cep.nfa.StateTransitionAction;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.GroupPattern;
import org.apache.flink.cep.pattern.MalformedPatternException;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.Quantifier;
import org.apache.flink.cep.pattern.Quantifier.Times;
import org.apache.flink.cep.pattern.WithinType;
import org.apache.flink.cep.pattern.conditions.BooleanConditions;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.RichAndCondition;
import org.apache.flink.cep.pattern.conditions.RichNotCondition;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.Stack;

import static org.apache.flink.cep.nfa.compiler.NFAStateNameHandler.STATE_NAME_DELIM;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * Compiler class containing methods to compile a {@link Pattern} into a {@link NFA} or a {@link
 * NFAFactory}.
 */
public class NFACompiler {

    protected static final String ENDING_STATE_NAME = "$endState$";

    /**
     * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
     * multiple NFAs.
     *
     * @param pattern Definition of sequence pattern
     * @param timeoutHandling True if the NFA shall return timed out event patterns
     * @param <T> Type of the input events
     * @return Factory for NFAs corresponding to the given pattern
     */
    @SuppressWarnings("unchecked")
    public static <T> NFAFactory<T> compileFactory(
            final Pattern<T, ?> pattern, boolean timeoutHandling) {
        if (pattern == null) {
            // return a factory for empty NFAs
            return new NFAFactoryImpl<>(
                    0,
                    Collections.<String, Long>emptyMap(),
                    Collections.<State<T>>emptyList(),
                    timeoutHandling);
        } else {
            final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
            nfaFactoryCompiler.compileFactory();
            return new NFAFactoryImpl<>(
                    nfaFactoryCompiler.getWindowTime(),
                    nfaFactoryCompiler.getWindowTimes(),
                    nfaFactoryCompiler.getStates(),
                    timeoutHandling);
        }
    }

    /**
     * Verifies if the provided pattern can possibly generate empty match. Example of patterns that
     * can possibly generate empty matches are: A*, A?, A* B? etc.
     *
     * @param pattern pattern to check
     * @return true if empty match could potentially match the pattern, false otherwise
     */
    public static boolean canProduceEmptyMatches(final Pattern<?, ?> pattern) {
        NFAFactoryCompiler<?> compiler = new NFAFactoryCompiler<>(checkNotNull(pattern));
        compiler.compileFactory();
        State<?> startState =
                compiler.getStates().stream()
                        .filter(State::isStart)
                        .findFirst()
                        .orElseThrow(
                                () ->
                                        new IllegalStateException(
                                                "Compiler produced no start state. It is a bug. File a jira."));

        Set<State<?>> visitedStates = new HashSet<>();
        final Stack<State<?>> statesToCheck = new Stack<>();
        statesToCheck.push(startState);
        while (!statesToCheck.isEmpty()) {
            final State<?> currentState = statesToCheck.pop();
            if (visitedStates.contains(currentState)) {
                continue;
            } else {
                visitedStates.add(currentState);
            }

            for (StateTransition<?> transition : currentState.getStateTransitions()) {
                if (transition.getAction() == StateTransitionAction.PROCEED) {
                    if (transition.getTargetState().isFinal()) {
                        return true;
                    } else {
                        statesToCheck.push(transition.getTargetState());
                    }
                }
            }
        }

        return false;
    }

    /**
     * Converts a {@link Pattern} into graph of {@link State}. It enables sharing of compilation
     * state across methods.
     *
     * @param <T>
     */
    static class NFAFactoryCompiler<T> {

        private final NFAStateNameHandler stateNameHandler = new NFAStateNameHandler();
        private final Map<String, State<T>> stopStates = new HashMap<>();
        private final List<State<T>> states = new ArrayList<>();
        private final Map<String, Long> windowTimes = new HashMap<>();

        private Optional<Long> windowTime;
        private GroupPattern<T, ?> currentGroupPattern;
        private Map<GroupPattern<T, ?>, Boolean> firstOfLoopMap = new HashMap<>();
        private Pattern<T, ?> currentPattern;
        private Pattern<T, ?> followingPattern;
        private final AfterMatchSkipStrategy afterMatchSkipStrategy;
        private Map<String, State<T>> originalStateMap = new HashMap<>();

        NFAFactoryCompiler(final Pattern<T, ?> pattern) {
            this.currentPattern = pattern;
            afterMatchSkipStrategy = pattern.getAfterMatchSkipStrategy();
            windowTime = Optional.empty();
        }

        /**
         * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to
         * create multiple NFAs.
         */
        void compileFactory() {

            Pattern<T, ?> lastPattern = currentPattern;

            checkPatternNameUniqueness();

            checkPatternSkipStrategy();

            // we're traversing the pattern from the end to the beginning --> the first state is the
            // final state
            State<T> sinkState = createEndingState();
            // add all the normal states
            sinkState = createMiddleStates(sinkState);
            // add the beginning state
            createStartState(sinkState);

            // check the window times between events for pattern
            checkPatternWindowTimes();

            if (lastPattern.getQuantifier().getConsumingStrategy()
                            == Quantifier.ConsumingStrategy.NOT_FOLLOW
                    && (!windowTimes.containsKey(lastPattern.getName())
                            || windowTimes.get(lastPattern.getName()) <= 0)
                    && getWindowTime() == 0) {
                throw new MalformedPatternException(
                        "NotFollowedBy is not supported without windowTime as a last part of a Pattern!");
            }
        }

        AfterMatchSkipStrategy getAfterMatchSkipStrategy() {
            return afterMatchSkipStrategy;
        }

        List<State<T>> getStates() {
            return states;
        }

        long getWindowTime() {
            return windowTime.orElse(0L);
        }

        Map<String, Long> getWindowTimes() {
            return windowTimes;
        }

        /** Check pattern window times between events. */
        private void checkPatternWindowTimes() {
            windowTime.ifPresent(
                    windowTime -> {
                        if (windowTimes.values().stream().anyMatch(time -> time > windowTime)) {
                            throw new MalformedPatternException(
                                    "The window length between the previous and current event cannot be larger than the window length between the first and last event for a Pattern.");
                        }
                    });
        }

        /** Check pattern after match skip strategy. */
        private void checkPatternSkipStrategy() {
            if (afterMatchSkipStrategy.getPatternName().isPresent()) {
                String patternName = afterMatchSkipStrategy.getPatternName().get();
                Pattern<T, ?> pattern = currentPattern;
                while (pattern.getPrevious() != null && !pattern.getName().equals(patternName)) {
                    pattern = pattern.getPrevious();
                }

                // pattern name match check.
                if (!pattern.getName().equals(patternName)) {
                    throw new MalformedPatternException(
                            "The pattern name specified in AfterMatchSkipStrategy "
                                    + "can not be found in the given Pattern");
                }
            }
        }

        /**
         * Check if there are duplicate pattern names. If yes, it throws a {@link
         * MalformedPatternException}.
         */
        private void checkPatternNameUniqueness() {
            // make sure there is no pattern with name "$endState$"
            stateNameHandler.checkNameUniqueness(ENDING_STATE_NAME);
            Pattern patternToCheck = currentPattern;
            while (patternToCheck != null) {
                checkPatternNameUniqueness(patternToCheck);
                patternToCheck = patternToCheck.getPrevious();
            }
            stateNameHandler.clear();
        }

        /**
         * Check if the given pattern's name is already used or not. If yes, it throws a {@link
         * MalformedPatternException}.
         *
         * @param pattern The pattern to be checked
         */
        private void checkPatternNameUniqueness(final Pattern pattern) {
            if (pattern instanceof GroupPattern) {
                Pattern patternToCheck = ((GroupPattern) pattern).getRawPattern();
                while (patternToCheck != null) {
                    checkPatternNameUniqueness(patternToCheck);
                    patternToCheck = patternToCheck.getPrevious();
                }
            } else {
                stateNameHandler.checkNameUniqueness(pattern.getName());
            }
        }

        /**
         * Retrieves list of conditions resulting in Stop state and names of the corresponding NOT
         * patterns.
         *
         * <p>A current not condition can be produced in two cases:
         *
         * <ol>
         *   <li>the previous pattern is a {@link Quantifier.ConsumingStrategy#NOT_FOLLOW}
         *   <li>exists a backward path of {@link Quantifier.QuantifierProperty#OPTIONAL} patterns
         *       to {@link Quantifier.ConsumingStrategy#NOT_FOLLOW}
         * </ol>
         *
         * <p><b>WARNING:</b> for more info on the second case see: {@link
         * NFAFactoryCompiler#copyWithoutTransitiveNots(State)}
         *
         * @return list of not conditions with corresponding names
         */
        private List<Tuple2<IterativeCondition<T>, String>> getCurrentNotCondition() {
            List<Tuple2<IterativeCondition<T>, String>> notConditions = new ArrayList<>();

            Pattern<T, ? extends T> previousPattern = currentPattern;
            while (previousPattern.getPrevious() != null
                    && (previousPattern
                                    .getPrevious()
                                    .getQuantifier()
                                    .hasProperty(Quantifier.QuantifierProperty.OPTIONAL)
                            || previousPattern.getPrevious().getQuantifier().getConsumingStrategy()
                                    == Quantifier.ConsumingStrategy.NOT_FOLLOW)) {

                previousPattern = previousPattern.getPrevious();

                if (previousPattern.getQuantifier().getConsumingStrategy()
                        == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
                    final IterativeCondition<T> notCondition = getTakeCondition(previousPattern);
                    notConditions.add(Tuple2.of(notCondition, previousPattern.getName()));
                }
            }
            return notConditions;
        }

        /**
         * Creates the dummy Final {@link State} of the NFA graph.
         *
         * @return dummy Final state
         */
        private State<T> createEndingState() {
            State<T> endState = createState(ENDING_STATE_NAME, State.StateType.Final);
            windowTime =
                    Optional.ofNullable(currentPattern.getWindowTime()).map(Time::toMilliseconds);
            return endState;
        }

        /**
         * Creates all the states between Start and Final state.
         *
         * @param sinkState the state that last state should point to (always the Final state)
         * @return the next state after Start in the resulting graph
         */
        private State<T> createMiddleStates(final State<T> sinkState) {
            State<T> lastSink = sinkState;
            while (currentPattern.getPrevious() != null) {

                if (currentPattern.getQuantifier().getConsumingStrategy()
                        == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
                    // skip notFollow patterns, they are converted into edge conditions
                    if ((currentPattern.getWindowTime(WithinType.PREVIOUS_AND_CURRENT) != null
                                    || getWindowTime() > 0)
                            && lastSink.isFinal()) {
                        final State<T> notFollow = createState(State.StateType.Pending, true);
                        final IterativeCondition<T> notCondition = getTakeCondition(currentPattern);
                        final State<T> stopState =
                                createStopState(notCondition, currentPattern.getName());
                        notFollow.addProceed(stopState, notCondition);
                        notFollow.addIgnore(new RichNotCondition<>(notCondition));
                        lastSink = notFollow;
                    }
                } else if (currentPattern.getQuantifier().getConsumingStrategy()
                        == Quantifier.ConsumingStrategy.NOT_NEXT) {
                    final State<T> notNext = createState(State.StateType.Normal, true);
                    final IterativeCondition<T> notCondition = getTakeCondition(currentPattern);
                    final State<T> stopState =
                            createStopState(notCondition, currentPattern.getName());

                    if (lastSink.isFinal()) {
                        // so that the proceed to final is not fired
                        notNext.addIgnore(lastSink, new RichNotCondition<>(notCondition));
                    } else {
                        notNext.addProceed(lastSink, new RichNotCondition<>(notCondition));
                    }
                    notNext.addProceed(stopState, notCondition);
                    lastSink = notNext;
                } else {
                    lastSink = convertPattern(lastSink);
                }

                // we traverse the pattern graph backwards
                followingPattern = currentPattern;
                currentPattern = currentPattern.getPrevious();

                final Time currentWindowTime = currentPattern.getWindowTime();
                if (currentWindowTime != null
                        && currentWindowTime.toMilliseconds() < windowTime.orElse(Long.MAX_VALUE)) {
                    // the window time is the global minimum of all window times of each state
                    windowTime = Optional.of(currentWindowTime.toMilliseconds());
                }
            }
            return lastSink;
        }

        /**
         * Creates the Start {@link State} of the resulting NFA graph.
         *
         * @param sinkState the state that Start state should point to (always first state of middle
         *     states)
         * @return created state
         */
        @SuppressWarnings("unchecked")
        private State<T> createStartState(State<T> sinkState) {
            final State<T> beginningState = convertPattern(sinkState);
            beginningState.makeStart();
            return beginningState;
        }

        private State<T> convertPattern(final State<T> sinkState) {
            final State<T> lastSink;

            final Quantifier quantifier = currentPattern.getQuantifier();
            if (quantifier.hasProperty(Quantifier.QuantifierProperty.LOOPING)) {

                // if loop has started then all notPatterns previous to the optional states are no
                // longer valid
                setCurrentGroupPatternFirstOfLoop(false);
                final State<T> sink = copyWithoutTransitiveNots(sinkState);
                final State<T> looping = createLooping(sink);

                setCurrentGroupPatternFirstOfLoop(true);
                lastSink = createTimesState(looping, sinkState, currentPattern.getTimes());
            } else if (quantifier.hasProperty(Quantifier.QuantifierProperty.TIMES)) {
                lastSink = createTimesState(sinkState, sinkState, currentPattern.getTimes());
            } else {
                lastSink = createSingletonState(sinkState);
            }
            addStopStates(lastSink);

            return lastSink;
        }

        private State<T> createState(State.StateType stateType, boolean isTake) {
            State<T> state = createState(currentPattern.getName(), stateType);
            if (isTake) {
                Times times = currentPattern.getTimes();
                Time windowTime = currentPattern.getWindowTime(WithinType.PREVIOUS_AND_CURRENT);
                if (times == null && windowTime != null) {
                    windowTimes.put(state.getName(), windowTime.toMilliseconds());
                } else if (times != null
                        && times.getWindowTime() != null
                        && state.getName().contains(STATE_NAME_DELIM)) {
                    windowTimes.put(state.getName(), times.getWindowTime().toMilliseconds());
                }
            }
            return state;
        }

        /**
         * Creates a state with {@link State.StateType#Normal} and adds it to the collection of
         * created states. Should be used instead of instantiating with new operator.
         *
         * @param name the name of the state
         * @param stateType the type of the state
         * @return the created state
         */
        private State<T> createState(String name, State.StateType stateType) {
            String stateName = stateNameHandler.getUniqueInternalName(name);
            State<T> state = new State<>(stateName, stateType);
            states.add(state);
            return state;
        }

        private State<T> createStopState(
                final IterativeCondition<T> notCondition, final String name) {
            // We should not duplicate the notStates. All states from which we can stop should point
            // to the same one.
            State<T> stopState = stopStates.get(name);
            if (stopState == null) {
                stopState = createState(name, State.StateType.Stop);
                stopState.addTake(notCondition);
                stopStates.put(name, stopState);
            }
            return stopState;
        }

        /**
         * This method creates an alternative state that is target for TAKE transition from an
         * optional State. Accepting an event in optional State discards all not Patterns that were
         * present before it.
         *
         * <p>E.g for a Pattern
         * begin("a").notFollowedBy("b").followedByAny("c").optional().followedByAny("d") a sequence
         * like : {a c b d} is a valid match, but {a b d} is not.
         *
         * <p><b>NOTICE:</b> This method creates copy only if it necessary.
         *
         * @param sinkState a state to create copy without transitive nots
         * @return the copy of the state itself if no modifications were needed
         */
        private State<T> copyWithoutTransitiveNots(final State<T> sinkState) {
            final List<Tuple2<IterativeCondition<T>, String>> currentNotCondition =
                    getCurrentNotCondition();

            if (currentNotCondition.isEmpty()
                    || !currentPattern
                            .getQuantifier()
                            .hasProperty(Quantifier.QuantifierProperty.OPTIONAL)) {
                // we do not create an alternative path if we are NOT in an OPTIONAL state or there
                // is no NOTs prior to
                // the optional state
                return sinkState;
            }

            final State<T> copyOfSink = createState(sinkState.getName(), sinkState.getStateType());

            for (StateTransition<T> tStateTransition : sinkState.getStateTransitions()) {

                if (tStateTransition.getAction() == StateTransitionAction.PROCEED) {
                    State<T> targetState = tStateTransition.getTargetState();
                    boolean remove = false;
                    if (targetState.isStop()) {
                        for (Tuple2<IterativeCondition<T>, String> notCondition :
                                currentNotCondition) {
                            if (targetState.getName().equals(notCondition.f1)) {
                                remove = true;
                            }
                        }
                    } else {
                        targetState = copyWithoutTransitiveNots(tStateTransition.getTargetState());
                    }

                    if (!remove) {
                        copyOfSink.addStateTransition(
                                tStateTransition.getAction(),
                                targetState,
                                tStateTransition.getCondition());
                    }
                } else {
                    copyOfSink.addStateTransition(
                            tStateTransition.getAction(),
                            tStateTransition
                                            .getTargetState()
                                            .equals(tStateTransition.getSourceState())
                                    ? copyOfSink
                                    : tStateTransition.getTargetState(),
                            tStateTransition.getCondition());
                }
            }
            return copyOfSink;
        }

        private State<T> copy(final State<T> state) {
            final State<T> copyOfState =
                    createState(
                            NFAStateNameHandler.getOriginalNameFromInternal(state.getName()),
                            state.getStateType());
            for (StateTransition<T> tStateTransition : state.getStateTransitions()) {
                copyOfState.addStateTransition(
                        tStateTransition.getAction(),
                        tStateTransition.getTargetState().equals(tStateTransition.getSourceState())
                                ? copyOfState
                                : tStateTransition.getTargetState(),
                        tStateTransition.getCondition());
            }
            return copyOfState;
        }

        private void addStopStates(final State<T> state) {
            for (Tuple2<IterativeCondition<T>, String> notCondition : getCurrentNotCondition()) {
                final State<T> stopState = createStopState(notCondition.f0, notCondition.f1);
                state.addProceed(stopState, notCondition.f0);
            }
        }

        private void addStopStateToLooping(final State<T> loopingState) {
            if (followingPattern != null
                    && followingPattern.getQuantifier().getConsumingStrategy()
                            == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
                final IterativeCondition<T> notCondition = getTakeCondition(followingPattern);
                final State<T> stopState =
                        createStopState(notCondition, followingPattern.getName());
                loopingState.addProceed(stopState, notCondition);
            }
        }

        /**
         * Creates a "complex" state consisting of given number of states with same {@link
         * IterativeCondition}.
         *
         * @param sinkState the state that the created state should point to
         * @param proceedState state that the state being converted should proceed to
         * @param times number of times the state should be copied
         * @return the first state of the "complex" state, next state should point to it
         */
        @SuppressWarnings("unchecked")
        private State<T> createTimesState(
                final State<T> sinkState, final State<T> proceedState, Times times) {
            State<T> lastSink = sinkState;
            setCurrentGroupPatternFirstOfLoop(false);
            final IterativeCondition<T> untilCondition =
                    (IterativeCondition<T>) currentPattern.getUntilCondition();
            final IterativeCondition<T> innerIgnoreCondition =
                    extendWithUntilCondition(
                            getInnerIgnoreCondition(currentPattern), untilCondition, false);
            final IterativeCondition<T> takeCondition =
                    extendWithUntilCondition(
                            getTakeCondition(currentPattern), untilCondition, true);

            if (currentPattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.GREEDY)
                    && times.getFrom() != times.getTo()) {
                if (untilCondition != null) {
                    State<T> sinkStateCopy = copy(sinkState);
                    originalStateMap.put(sinkState.getName(), sinkStateCopy);
                }
                updateWithGreedyCondition(sinkState, takeCondition);
            }

            for (int i = times.getFrom(); i < times.getTo(); i++) {
                lastSink =
                        createSingletonState(
                                lastSink, proceedState, takeCondition, innerIgnoreCondition, true);
                addStopStateToLooping(lastSink);
            }
            for (int i = 0; i < times.getFrom() - 1; i++) {
                lastSink =
                        createSingletonState(
                                lastSink, null, takeCondition, innerIgnoreCondition, false);
                addStopStateToLooping(lastSink);
            }
            // we created the intermediate states in the loop, now we create the start of the loop.
            setCurrentGroupPatternFirstOfLoop(true);
            return createSingletonState(
                    lastSink,
                    proceedState,
                    takeCondition,
                    getIgnoreCondition(currentPattern),
                    isPatternOptional(currentPattern));
        }

        /**
         * Marks the current group pattern as the head of the TIMES quantifier or not.
         *
         * @param isFirstOfLoop whether the current group pattern is the head of the TIMES
         *     quantifier
         */
        @SuppressWarnings("unchecked")
        private void setCurrentGroupPatternFirstOfLoop(boolean isFirstOfLoop) {
            if (currentPattern instanceof GroupPattern) {
                firstOfLoopMap.put((GroupPattern<T, ?>) currentPattern, isFirstOfLoop);
            }
        }

        /**
         * Checks if the current group pattern is the head of the TIMES/LOOPING quantifier or not a
         * TIMES/LOOPING quantifier pattern.
         */
        private boolean isCurrentGroupPatternFirstOfLoop() {
            if (firstOfLoopMap.containsKey(currentGroupPattern)) {
                return firstOfLoopMap.get(currentGroupPattern);
            } else {
                return true;
            }
        }

        /**
         * Checks if the given pattern is the head pattern of the current group pattern.
         *
         * @param pattern the pattern to be checked
         * @return {@code true} iff the given pattern is in a group pattern and it is the head
         *     pattern of the group pattern, {@code false} otherwise
         */
        private boolean headOfGroup(Pattern<T, ?> pattern) {
            return currentGroupPattern != null && pattern.getPrevious() == null;
        }

        /**
         * Checks if the given pattern is optional. If the given pattern is the head of a group
         * pattern, the optional status depends on the group pattern.
         */
        private boolean isPatternOptional(Pattern<T, ?> pattern) {
            return pattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL);
        }

        private boolean isHeadOfOptionalGroupPattern(Pattern<T, ?> pattern) {
            if (!headOfGroup(pattern)) {
                return false;
            }
            return isCurrentGroupPatternFirstOfLoop()
                    && currentGroupPattern
                            .getQuantifier()
                            .hasProperty(Quantifier.QuantifierProperty.OPTIONAL);
        }

        /**
         * Creates a simple single state. For an OPTIONAL state it also consists of a similar state
         * without the PROCEED edge, so that for each PROCEED transition branches in computation
         * state graph can be created only once.
         *
         * @param sinkState state that the state being converted should point to
         * @return the created state
         */
        @SuppressWarnings("unchecked")
        private State<T> createSingletonState(final State<T> sinkState) {
            return createSingletonState(
                    sinkState,
                    sinkState,
                    getTakeCondition(currentPattern),
                    getIgnoreCondition(currentPattern),
                    isPatternOptional(currentPattern));
        }

        /**
         * Creates a simple single state. For an OPTIONAL state it also consists of a similar state
         * without the PROCEED edge, so that for each PROCEED transition branches in computation
         * state graph can be created only once.
         *
         * @param ignoreCondition condition that should be applied to IGNORE transition
         * @param sinkState state that the state being converted should point to
         * @param proceedState state that the state being converted should proceed to
         * @param isOptional whether the state being converted is optional
         * @return the created state
         */
        @SuppressWarnings("unchecked")
        private State<T> createSingletonState(
                final State<T> sinkState,
                final State<T> proceedState,
                final IterativeCondition<T> takeCondition,
                final IterativeCondition<T> ignoreCondition,
                final boolean isOptional) {
            if (currentPattern instanceof GroupPattern) {
                return createGroupPatternState(
                        (GroupPattern) currentPattern, sinkState, proceedState, isOptional);
            }

            final State<T> singletonState = createState(State.StateType.Normal, true);
            // if event is accepted then all notPatterns previous to the optional states are no
            // longer valid
            final State<T> sink = copyWithoutTransitiveNots(sinkState);
            singletonState.addTake(sink, takeCondition);

            // if no element accepted the previous nots are still valid.
            final IterativeCondition<T> proceedCondition = getTrueFunction();

            if (isOptional) {
                if (currentPattern
                        .getQuantifier()
                        .hasProperty(Quantifier.QuantifierProperty.GREEDY)) {
                    final IterativeCondition<T> untilCondition =
                            (IterativeCondition<T>) currentPattern.getUntilCondition();
                    if (untilCondition != null) {
                        singletonState.addProceed(
                                originalStateMap.get(proceedState.getName()),
                                new RichAndCondition<>(proceedCondition, untilCondition));
                    }
                    singletonState.addProceed(
                            proceedState,
                            untilCondition != null
                                    ? new RichAndCondition<>(
                                            proceedCondition,
                                            new RichNotCondition<>(untilCondition))
                                    : proceedCondition);
                } else {
                    singletonState.addProceed(proceedState, proceedCondition);
                }
            }

            if (ignoreCondition != null) {
                final State<T> ignoreState;
                if (isOptional || isHeadOfOptionalGroupPattern(currentPattern)) {
                    ignoreState = createState(State.StateType.Normal, false);
                    ignoreState.addTake(sink, takeCondition);
                    ignoreState.addIgnore(ignoreCondition);
                    addStopStates(ignoreState);
                } else {
                    ignoreState = singletonState;
                }
                singletonState.addIgnore(ignoreState, ignoreCondition);
            }
            return singletonState;
        }

        /**
         * Create all the states for the group pattern.
         *
         * @param groupPattern the group pattern to create the states for
         * @param sinkState the state that the group pattern being converted should point to
         * @param proceedState the state that the group pattern being converted should proceed to
         * @param isOptional whether the group pattern being converted is optional
         * @return the first state of the states of the group pattern
         */
        private State<T> createGroupPatternState(
                final GroupPattern<T, ?> groupPattern,
                final State<T> sinkState,
                final State<T> proceedState,
                final boolean isOptional) {
            final IterativeCondition<T> proceedCondition = getTrueFunction();

            Pattern<T, ?> oldCurrentPattern = currentPattern;
            Pattern<T, ?> oldFollowingPattern = followingPattern;
            GroupPattern<T, ?> oldGroupPattern = currentGroupPattern;

            State<T> lastSink = sinkState;
            currentGroupPattern = groupPattern;
            currentPattern = groupPattern.getRawPattern();
            lastSink = createMiddleStates(lastSink);
            lastSink = convertPattern(lastSink);
            if (isOptional) {
                // for the first state of a group pattern, its PROCEED edge should point to
                // the following state of that group pattern
                lastSink.addProceed(proceedState, proceedCondition);
            }
            currentPattern = oldCurrentPattern;
            followingPattern = oldFollowingPattern;
            currentGroupPattern = oldGroupPattern;
            return lastSink;
        }

        /**
         * Create the states for the group pattern as a looping one.
         *
         * @param groupPattern the group pattern to create the states for
         * @param sinkState the state that the group pattern being converted should point to
         * @return the first state of the states of the group pattern
         */
        private State<T> createLoopingGroupPatternState(
                final GroupPattern<T, ?> groupPattern, final State<T> sinkState) {
            final IterativeCondition<T> proceedCondition = getTrueFunction();

            Pattern<T, ?> oldCurrentPattern = currentPattern;
            Pattern<T, ?> oldFollowingPattern = followingPattern;
            GroupPattern<T, ?> oldGroupPattern = currentGroupPattern;

            final State<T> dummyState = createState(State.StateType.Normal, true);
            State<T> lastSink = dummyState;
            currentGroupPattern = groupPattern;
            currentPattern = groupPattern.getRawPattern();
            lastSink = createMiddleStates(lastSink);
            lastSink = convertPattern(lastSink);
            lastSink.addProceed(sinkState, proceedCondition);
            dummyState.addProceed(lastSink, proceedCondition);
            currentPattern = oldCurrentPattern;
            followingPattern = oldFollowingPattern;
            currentGroupPattern = oldGroupPattern;
            return lastSink;
        }

        /**
         * Creates the given state as a looping one. Looping state is one with TAKE edge to itself
         * and PROCEED edge to the sinkState. It also consists of a similar state without the
         * PROCEED edge, so that for each PROCEED transition branches in computation state graph can
         * be created only once.
         *
         * @param sinkState the state that the converted state should point to
         * @return the first state of the created complex state
         */
        @SuppressWarnings("unchecked")
        private State<T> createLooping(final State<T> sinkState) {
            if (currentPattern instanceof GroupPattern) {
                return createLoopingGroupPatternState((GroupPattern) currentPattern, sinkState);
            }
            final IterativeCondition<T> untilCondition =
                    (IterativeCondition<T>) currentPattern.getUntilCondition();

            final IterativeCondition<T> ignoreCondition =
                    extendWithUntilCondition(
                            getInnerIgnoreCondition(currentPattern), untilCondition, false);
            final IterativeCondition<T> takeCondition =
                    extendWithUntilCondition(
                            getTakeCondition(currentPattern), untilCondition, true);

            IterativeCondition<T> proceedCondition = getTrueFunction();
            final State<T> loopingState = createState(State.StateType.Normal, true);

            if (currentPattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.GREEDY)) {
                if (untilCondition != null) {
                    State<T> sinkStateCopy = copy(sinkState);
                    loopingState.addProceed(
                            sinkStateCopy,
                            new RichAndCondition<>(proceedCondition, untilCondition));
                    originalStateMap.put(sinkState.getName(), sinkStateCopy);
                }
                loopingState.addProceed(
                        sinkState,
                        untilCondition != null
                                ? new RichAndCondition<>(
                                        proceedCondition, new RichNotCondition<>(untilCondition))
                                : proceedCondition);
                updateWithGreedyCondition(sinkState, getTakeCondition(currentPattern));
            } else {
                loopingState.addProceed(sinkState, proceedCondition);
            }
            loopingState.addTake(takeCondition);

            addStopStateToLooping(loopingState);

            if (ignoreCondition != null) {
                final State<T> ignoreState = createState(State.StateType.Normal, false);
                ignoreState.addTake(loopingState, takeCondition);
                ignoreState.addIgnore(ignoreCondition);
                loopingState.addIgnore(ignoreState, ignoreCondition);

                addStopStateToLooping(ignoreState);
            }
            return loopingState;
        }

        /**
         * This method extends the given condition with stop(until) condition if necessary. The
         * until condition needs to be applied only if both of the given conditions are not null.
         *
         * @param condition the condition to extend
         * @param untilCondition the until condition to join with the given condition
         * @param isTakeCondition whether the {@code condition} is for {@code TAKE} edge
         * @return condition with AND applied or the original condition
         */
        private IterativeCondition<T> extendWithUntilCondition(
                IterativeCondition<T> condition,
                IterativeCondition<T> untilCondition,
                boolean isTakeCondition) {
            if (untilCondition != null && condition != null) {
                return new RichAndCondition<>(new RichNotCondition<>(untilCondition), condition);
            } else if (untilCondition != null && isTakeCondition) {
                return new RichNotCondition<>(untilCondition);
            }

            return condition;
        }

        /**
         * @return The {@link IterativeCondition condition} for the {@code IGNORE} edge that
         *     corresponds to the specified {@link Pattern} and extended with stop(until) condition
         *     if necessary. It is applicable only for inner states of a complex state like looping
         *     or times.
         */
        @SuppressWarnings("unchecked")
        private IterativeCondition<T> getInnerIgnoreCondition(Pattern<T, ?> pattern) {
            Quantifier.ConsumingStrategy consumingStrategy =
                    pattern.getQuantifier().getInnerConsumingStrategy();
            if (headOfGroup(pattern)) {
                // for the head pattern of a group pattern, we should consider the
                // inner consume strategy of the group pattern
                consumingStrategy = currentGroupPattern.getQuantifier().getInnerConsumingStrategy();
            }

            IterativeCondition<T> innerIgnoreCondition = null;
            switch (consumingStrategy) {
                case STRICT:
                    innerIgnoreCondition = null;
                    break;
                case SKIP_TILL_NEXT:
                    innerIgnoreCondition =
                            new RichNotCondition<>((IterativeCondition<T>) pattern.getCondition());
                    break;
                case SKIP_TILL_ANY:
                    innerIgnoreCondition = BooleanConditions.trueFunction();
                    break;
            }

            if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
                innerIgnoreCondition =
                        extendWithUntilCondition(
                                innerIgnoreCondition,
                                (IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
                                false);
            }
            return innerIgnoreCondition;
        }

        /**
         * @return The {@link IterativeCondition condition} for the {@code IGNORE} edge that
         *     corresponds to the specified {@link Pattern} and extended with stop(until) condition
         *     if necessary. For more on strategy see {@link Quantifier}
         */
        @SuppressWarnings("unchecked")
        private IterativeCondition<T> getIgnoreCondition(Pattern<T, ?> pattern) {
            Quantifier.ConsumingStrategy consumingStrategy =
                    pattern.getQuantifier().getConsumingStrategy();
            if (headOfGroup(pattern)) {
                // for the head pattern of a group pattern, we should consider the inner consume
                // strategy
                // of the group pattern if the group pattern is not the head of the TIMES/LOOPING
                // quantifier;
                // otherwise, we should consider the consume strategy of the group pattern
                if (isCurrentGroupPatternFirstOfLoop()) {
                    consumingStrategy = currentGroupPattern.getQuantifier().getConsumingStrategy();
                } else {
                    consumingStrategy =
                            currentGroupPattern.getQuantifier().getInnerConsumingStrategy();
                }
            }

            IterativeCondition<T> ignoreCondition = null;
            switch (consumingStrategy) {
                case STRICT:
                    ignoreCondition = null;
                    break;
                case SKIP_TILL_NEXT:
                    ignoreCondition =
                            new RichNotCondition<>((IterativeCondition<T>) pattern.getCondition());
                    break;
                case SKIP_TILL_ANY:
                    ignoreCondition = BooleanConditions.trueFunction();
                    break;
            }

            if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
                ignoreCondition =
                        extendWithUntilCondition(
                                ignoreCondition,
                                (IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
                                false);
            }
            return ignoreCondition;
        }

        /**
         * @return the {@link IterativeCondition condition} for the {@code TAKE} edge that
         *     corresponds to the specified {@link Pattern} and extended with stop(until) condition
         *     if necessary.
         */
        @SuppressWarnings("unchecked")
        private IterativeCondition<T> getTakeCondition(Pattern<T, ?> pattern) {
            IterativeCondition<T> takeCondition = (IterativeCondition<T>) pattern.getCondition();
            if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
                takeCondition =
                        extendWithUntilCondition(
                                takeCondition,
                                (IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
                                true);
            }
            return takeCondition;
        }

        /** @return An true function extended with stop(until) condition if necessary. */
        @SuppressWarnings("unchecked")
        private IterativeCondition<T> getTrueFunction() {
            IterativeCondition<T> trueCondition = BooleanConditions.trueFunction();
            if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
                trueCondition =
                        extendWithUntilCondition(
                                trueCondition,
                                (IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
                                true);
            }
            return trueCondition;
        }

        private void updateWithGreedyCondition(
                State<T> state, IterativeCondition<T> takeCondition) {
            for (StateTransition<T> stateTransition : state.getStateTransitions()) {
                stateTransition.setCondition(
                        new RichAndCondition<>(
                                stateTransition.getCondition(),
                                new RichNotCondition<>(takeCondition)));
            }
        }
    }

    /**
     * Factory interface for {@link NFA}.
     *
     * @param <T> Type of the input events which are processed by the NFA
     */
    public interface NFAFactory<T> extends Serializable {
        NFA<T> createNFA();
    }

    /**
     * Implementation of the {@link NFAFactory} interface.
     *
     * <p>The implementation takes the input type serializer, the window time and the set of states
     * and their transitions to be able to create an NFA from them.
     *
     * @param <T> Type of the input events which are processed by the NFA
     */
    private static class NFAFactoryImpl<T> implements NFAFactory<T> {

        private static final long serialVersionUID = 8939783698296714379L;

        private final long windowTime;
        private final Map<String, Long> windowTimes;
        private final Collection<State<T>> states;
        private final boolean timeoutHandling;

        private NFAFactoryImpl(
                long windowTime,
                Map<String, Long> windowTimes,
                Collection<State<T>> states,
                boolean timeoutHandling) {

            this.windowTime = windowTime;
            this.windowTimes = windowTimes;
            this.states = states;
            this.timeoutHandling = timeoutHandling;
        }

        @Override
        public NFA<T> createNFA() {
            return new NFA<>(states, windowTimes, windowTime, timeoutHandling);
        }
    }
}
