package cz.senslog.analytics.module;

import cz.senslog.analytics.domain.*;
import cz.senslog.analytics.repository.AnalyticsRepository;
import cz.senslog.analytics.utils.newvalidator.NotifyTrigger;
import cz.senslog.analytics.utils.newvalidator.ThresholdChecker;
import cz.senslog.analytics.module.api.Module;
import cz.senslog.analytics.module.api.ModuleDescriptor;
import cz.senslog.analytics.module.api.SimpleModule;
import cz.senslog.analytics.utils.newvalidator.Validator;
import cz.senslog.analytics.utils.Tuple;
import cz.senslog.analytics.utils.newvalidator.ViolationReport;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Collections.emptyMap;

public class ObservationReceiverModule extends SimpleModule {

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

    private record RawSensor(long unitId, long sensorId) {}

    private static final String MODULE_NAME = "RECEIVER";

    private final AnalyticsRepository repo;

    private Map<RawSensor, Sensor> sensorsMap;
    private Map<Long, Sensor> sensorIdToSensorMap;
    private Map<Long, List<AnalyticGroup>> sensorToMultiGroup;
    private ThresholdChecker<Observation> thresholdChecker;

    public ObservationReceiverModule(AnalyticsRepository repo) {
        this.repo = repo;
        this.sensorsMap = emptyMap();
        this.sensorToMultiGroup = emptyMap();
        this.thresholdChecker = ThresholdChecker.disabled();
    }

    @Override
    public void configure(Promise<Void> completePromise) {

        Future<List<Sensor>> allSensorsFuture = repo.loadAllSensors();
        Future<List<Threshold>> thresholdsForSensorsFuture = repo.loadThresholdsForSensor();
        Future<List<Tuple<Long, AnalyticGroup>>> datasourceToAnalyticGroupFuture = repo.loadAllAnalyticGroupsWithDatasource();

        final Validator<Observation> validator = Validator.<Observation>create()
                .addMapping(AttributeType.VAL, o -> o::value)
                .addMapping(AttributeType.TIME, o -> () ->  (double) o.timestamp().toLocalTime().toSecondOfDay());

        Future.all(allSensorsFuture, thresholdsForSensorsFuture, datasourceToAnalyticGroupFuture)
                .onSuccess(ar -> {
                    List<Sensor> allSensors = ar.resultAt(0);
                    sensorsMap = new HashMap<>(allSensors.size());
                    sensorIdToSensorMap = new HashMap<>(allSensors.size());
                    for (Sensor s : allSensors) {
                        sensorsMap.put(new RawSensor(s.unitId(), s.sensorId()), s);
                        sensorIdToSensorMap.put(s.id(), s);
                    }

                    List<Threshold> sensorsThresholds = ar.resultAt(1);
                    // TODO refactor
                    thresholdChecker = ThresholdChecker.create(MODULE_NAME,
                            sensorsThresholds.stream().map(t -> new cz.senslog.analytics.utils.newvalidator.Threshold(t.id(), t.datasourceId(), t.enableProcess(), t.enableAlert(), t.customAlertMessage(),
                                    t.thresholdDimensions().stream().map(d -> new cz.senslog.analytics.utils.newvalidator.Threshold.Dimension(d.attributeType(), NotifyTrigger.Mode.of(d.notifyTriggerMode().name()),
                                            d.rules().stream().map(r -> new cz.senslog.analytics.utils.newvalidator.Threshold.Dimension.Rule(r.comparisonOperator(), r.value())).toList())).toList())).toList(),
                            validator, this::notifyIfViolated, true);

                    List<Tuple<Long, AnalyticGroup>> groupsWithDS = ar.resultAt(2);
                    sensorToMultiGroup = groupsWithDS.stream()
                            .filter(t -> Module.containsCollectorOf(t.getItem2().type()))
                            .collect(Collectors.groupingBy(Tuple::getItem1, TreeMap::new, Collectors.mapping(Tuple::getItem2, Collectors.toList())));

                    completePromise.complete();
                })
                .onFailure(completePromise::fail);
    }

    private void notifyIfViolated(ViolationReport report) {
        Sensor sensor = sensorIdToSensorMap.get(report.dataSourceId());
        String sourceName = String.format("unit(%d)/sensor(%d)", sensor.unitId(), sensor.sensorId());
        notify(ThresholdViolationAlert.of(MODULE_NAME, report.dataSourceId(), sourceName, report.message(), report.timestamp()));
    }

    @Override
    public void run(Promise<Void> completePromise) {
        eventBus().<ListHolder<RawObservation>>consumer(id(), h -> {
            logger.info("Received {} observations.", h.body().size());
            h.body().stream()
                            .filter(o -> sensorsMap.containsKey(new RawSensor(o.unitId(), o.sensorId())))
                            .map(o -> Observation.of(sensorsMap.get(new RawSensor(o.unitId(), o.sensorId())), o))
                            .peek(o -> repo.updateLastTimestamp(o.sensor().id(), o.timestamp()).onFailure(logger::error))
                            .filter(thresholdChecker::check)
                            .flatMap(this::mappingToGroups)
                            .forEach(m -> eventBus().publish(m.getItem1().id(), m.getItem2()));
            h.reply(String.format("Processed %d observations.", h.body().size()));
        }).completionHandler(ar -> {
            if (ar.succeeded()) {
                completePromise.complete();
            } else {
                completePromise.fail(ar.cause());
            }
        });
    }

    private Stream<Tuple<ModuleDescriptor, AnalyticObservation>> mappingToGroups(Observation o) {
        if (sensorToMultiGroup.containsKey(o.datasourceId())) {
            return sensorToMultiGroup.get(o.datasourceId()).stream()
                    .map(g -> Tuple.of(Module.collectorOf(g.type()), AnalyticObservation.of(o, g.id())));
        } else {
            return Stream.empty();
        }
    }
}