package cz.senslog.analytics.module.api;

import cz.senslog.analytics.domain.*;
import cz.senslog.analytics.repository.AnalyticsRepository;
import io.vertx.core.Promise;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.time.Instant;
import java.time.OffsetDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static cz.senslog.analytics.utils.TimeUtils.minToMillis;
import static cz.senslog.analytics.utils.TimeUtils.minToSec;

public abstract class CollectorModule extends SimpleModule {

    private static final Logger logger = LogManager.getLogger(CollectorModule.class);

    private static final int MAX_WAITING_TO_HARVEST_MIN = 10;

    private final CollectorType type;

    protected final AnalyticsRepository repo;

    private Map<Long, List<CollectedStatistics>> collectedStats;
    private Map<Long, AnalyticGroup> groupMap;
    private List<Threshold> thresholds;

    private Instant lastHarvesting;

    protected CollectorModule(CollectorType type, AnalyticsRepository repo) {
        this.type = type;
        this.repo = repo;

        this.groupMap = Collections.emptyMap();
        this.thresholds = Collections.emptyList();
        this.collectedStats = Collections.emptyMap();
    }

    protected abstract void handle(Collection<DoubleStatistics> aggregations);

    public final CollectorType type() {
        return type;
    }

    protected void init() {}

    @Override
    protected final void configure(Promise<Void> completePromise) {
        lastHarvesting = Instant.now();
        repo.loadGroupsByCollectorType(type).onSuccess(groups -> {
            collectedStats = new ConcurrentHashMap<>(groups.size());
            groupMap = groups;

            repo.loadGroupThresholds(type).onSuccess(data -> {
                thresholds = data;
                init();
                completePromise.complete();
            }).onFailure(completePromise::fail);
        }).onFailure(completePromise::fail);

        long period = minToMillis(MAX_WAITING_TO_HARVEST_MIN);
        vertx.setPeriodic(period, period, l -> {
           if (overtakeHarvestTime()) {
               harvestStatistics();
           }
        });
    }

    private void harvestStatistics() {
        OffsetDateTime currentHarvest = OffsetDateTime.now();
        for (List<CollectedStatistics> statistics : collectedStats.values()) {
            Iterator<CollectedStatistics> stIterator = statistics.iterator();
            while (stIterator.hasNext()) {
                CollectedStatistics st = stIterator.next();
                if (currentHarvest.isAfter(st.endTime())) {
                    handle(st.statistics());
                    stIterator.remove();
                }
            }
        }
        lastHarvesting = currentHarvest.toInstant();
    }

    private boolean overtakeHarvestTime() {
        return lastHarvesting.plusSeconds(minToSec(MAX_WAITING_TO_HARVEST_MIN)).isBefore(Instant.now());
    }

    protected List<Threshold> thresholds() {
        return thresholds;
    }

    protected Map<Long, AnalyticGroup> groups() {
        return groupMap;
    }

    private void collect(AnalyticObservation observation) {

        long datasourceId = observation.datasourceId();
        OffsetDateTime timestamp = observation.timestamp();
        double observedValue = observation.value();

        AnalyticGroup analyticGroup = groupMap.get(observation.analyticGroupId());

        List<CollectedStatistics> groupStatistics = collectedStats.computeIfAbsent(analyticGroup.id(), g -> new ArrayList<>());

        boolean newDataAccepted = false;
        Iterator<CollectedStatistics> statisticsIterator = groupStatistics.iterator();
        while(statisticsIterator.hasNext()) {
            CollectedStatistics st = statisticsIterator.next();
            if (timestamp.isEqual(st.startTime()) ||    // startInterval <= timestamp < endInterval
                    (timestamp.isAfter(st.startTime()) && timestamp.isBefore(st.endTime()))
            ) {
                st.accept(datasourceId, observedValue);
                newDataAccepted = true;
            } else if (timestamp.isAfter(st.endTime())) {
                handle(st.statistics());
                statisticsIterator.remove();
            }
        }
        lastHarvesting = Instant.now();

        if (!newDataAccepted) {
            CollectedStatistics newSt = CollectedStatistics.init(analyticGroup, timestamp);
            newSt.accept(datasourceId, observedValue);
            groupStatistics.add(newSt);
        }
    }

    @Override
    public void run(Promise<Void> completePromise) {
        eventBus().<AnalyticObservation>consumer(id(), v -> collect(v.body()))
                .completionHandler(ar -> {
                    if (ar.succeeded()) {
                        completePromise.complete();
                    } else {
                        completePromise.fail(ar.cause());
                    }
                });
    }

    protected final void persist(StatisticRecord record) {
        repo.saveStatisticRecord(record).onFailure(logger::catching);
    }
}