package com.groupon.lex.metrics.timeseries;

import com.groupon.lex.metrics.history.CollectHistory;
import com.groupon.lex.metrics.lib.ForwardIterator;
import java.util.ArrayList;
import static java.util.Collections.unmodifiableList;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.joda.time.DateTime;

/**
 * A TimeSeriesCollectionPair that keeps track of historical data.
 */
public abstract class AbstractTSCPair implements TimeSeriesCollectionPair {
    private static final Logger LOG = Logger.getLogger(AbstractTSCPair.class.getName());
    private List<TimeSeriesCollection> previous_;

    protected AbstractTSCPair() {
        previous_ = new ArrayList<>();
    }

    protected AbstractTSCPair(AbstractTSCPair original) {
        previous_ = new ArrayList<>(original.previous_);
    }

    private void validatePrevious() {
        try {
            DateTime ts = getCurrentCollection().getTimestamp();

            for (TimeSeriesCollection p : previous_) {
                if (!p.getTimestamp().isBefore(ts))
                    throw new IllegalArgumentException("previous timestamps must be before current and ordered in reverse chronological order");
                ts = p.getTimestamp();
            }
        } catch (IllegalArgumentException ex) {
            LOG.log(Level.SEVERE, "programmer error", ex);
            throw ex;
        }
    }

    @Override
    public abstract TimeSeriesCollection getCurrentCollection();

    public void initWithHistoricalData(CollectHistory history, ExpressionLookBack lookback) {
        if (!previous_.isEmpty()) {
            LOG.log(Level.WARNING, "skipping historical data initialization, as data is already present");
            return;
        }

        Stream<TimeSeriesCollection> filtered;
        try {
            filtered = lookback.filter(new ForwardIterator<>(history.streamReversed().iterator()));
        } catch (UnsupportedOperationException ex) {
            LOG.log(Level.WARNING, "history reverse streaming not supported, fallback to duration hint");
            final DateTime end = history.getEnd();
            final DateTime begin = end.minus(lookback.hintDuration());
            filtered = history.stream(begin, end);
        }

        for (TimeSeriesCollection tsc : filtered
                .sorted()
                .distinct()
                .collect(Collectors.toList())) {
            previous_.add(0, tsc);
        }
        LOG.log(Level.INFO, "recovered {0} scrapes from history", previous_.size());

        validatePrevious();  // Should never trigger.
    }

    @Override
    public Optional<TimeSeriesCollection> getPreviousCollection(int n) {
        if (n < 0) throw new IllegalArgumentException("cannot look into the future");
        if (n == 0) return Optional.of(getCurrentCollection());
        if (n - 1 >= previous_.size()) return Optional.empty();
        return Optional.of(previous_.get(n - 1));
    }

    protected final void update(TimeSeriesCollection tsc, ExpressionLookBack lookback, Runnable doBeforeValidation) {
        previous_.add(0, tsc);
        apply_lookback_(lookback);
        doBeforeValidation.run();

        validatePrevious();
    }

    private void apply_lookback_(ExpressionLookBack lookback) {
        final List<TimeSeriesCollection> suggested_previous = lookback.filter(new ForwardIterator<>(unmodifiableList(previous_).iterator()))
                .distinct()
                .sorted(Comparator.comparing(TimeSeriesCollection::getTimestamp).reversed())
                .collect(Collectors.toList());

        if (suggested_previous.isEmpty())  // Always keep at least 1 element.
            previous_.subList(1, previous_.size()).clear();
        else
            previous_ = suggested_previous;
    }

    @Override
    public String toString() {
        return previous_.stream()
                .map(TimeSeriesCollection::getTimestamp)
                .map(DateTime::toString)
                .collect(Collectors.joining(", ", "AbstractTSCPair[", "]"));
    }

    @Override
    public int size() {
        return previous_.size() + 1;
    }
}

@@ -42,63 +42,8 @@
 s;%0A%0A
-import com.groupon.lex.metrics.history.CollectHistory;%0A
 impo
@@ -356,40 +356,8 @@
 rs;%0A
-import java.util.stream.Stream;%0A
 impo
@@ -1530,1124 +1530,8 @@
 );%0A%0A
-    public void initWithHistoricalData(CollectHistory history, ExpressionLookBack lookback) %7B%0A        if (!previous_.isEmpty()) %7B%0A            LOG.log(Level.WARNING, %22skipping historical data initialization, as data is already present%22);%0A            return;%0A        %7D%0A%0A        Stream%3CTimeSeriesCollection%3E filtered;%0A        try %7B%0A            filtered = lookback.filter(new ForwardIterator%3C%3E(history.streamReversed().iterator()));%0A        %7D catch (UnsupportedOperationException ex) %7B%0A            LOG.log(Level.WARNING, %22history reverse streaming not supported, fallback to duration hint%22);%0A            final DateTime end = history.getEnd();%0A            final DateTime begin = end.minus(lookback.hintDuration());%0A            filtered = history.stream(begin, end);%0A        %7D%0A%0A        for (TimeSeriesCollection tsc : filtered%0A                .sorted()%0A                .distinct()%0A                .collect(Collectors.toList())) %7B%0A            previous_.add(0, tsc);%0A        %7D%0A        LOG.log(Level.INFO, %22recovered %7B0%7D scrapes from history%22, previous_.size());%0A%0A        validatePrevious();  // Should never trigger.%0A    %7D%0A%0A
     
@@ -1631,16 +1631,28 @@
  (n %3C 0)
+%0A           
  throw n
@@ -2565,17 +2565,16 @@
 Empty())
- 
  // Alwa

Remove logic to initialize AbstractTSCPair with historical data, since it's no longer used.