package ctbrec.recorder;

import static ctbrec.Model.State.*;

import java.io.InterruptedIOException;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.time.Instant;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ctbrec.Config;
import ctbrec.Model;
import ctbrec.event.EventBusHolder;
import ctbrec.event.ModelIsOnlineEvent;
import ctbrec.event.ModelStateChangedEvent;
import ctbrec.io.HttpException;

public class OnlineMonitor extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(OnlineMonitor.class);
    private static final boolean IGNORE_CACHE = true;

    private volatile boolean running = false;
    private Recorder recorder;

    private Map<Model, Model.State> states = new HashMap<>();

    private Map<String, ExecutorService> executors = new HashMap<>();

    public OnlineMonitor(Recorder recorder) {
        this.recorder = recorder;
        setName("OnlineMonitor");
        setDaemon(true);
    }

    @Override
    public void run() {
        running = true;
        while (running) {
            Instant begin = Instant.now();
            List<Model> models = recorder.getModels();

            // remove models, which are not recorded anymore
            removeDeletedModels(models);

            // update the currently recorded models
            updateModels(models);

            Instant end = Instant.now();
            Duration timeCheckTook = Duration.between(begin, end);
            suspendUntilNextIteration(models, timeCheckTook);
        }
        LOG.debug("{} terminated", getName());
    }

    private void removeDeletedModels(List<Model> models) {
        for (Iterator<Model> iterator = states.keySet().iterator(); iterator.hasNext();) {
            Model model = iterator.next();
            if(!models.contains(model)) {
                iterator.remove();
            }
        }
    }

    private void updateModels(List<Model> models) {
        // sort models by priority
        Collections.sort(models, (a, b) -> b.getPriority() - a.getPriority());
        // submit online check jobs to the executor for the model's site
        List<Future<?>> futures = new LinkedList<>();
        for (Model model : models) {
            futures.add(updateModel(model));
        }
        // wait for all jobs to finish
        for (Future<?> future : futures) {
            try {
                future.get();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (ExecutionException e) {
                LOG.info("Error while checking online state", e);
            }
        }
    }

    private Future<?> updateModel(Model model) {
        final String siteName = model.getSite().getName();
        ExecutorService executor = executors.computeIfAbsent(siteName, name -> Executors.newSingleThreadExecutor(r -> {
            Thread t = new Thread(r);
            t.setName("OnlineMonitorThread-" + siteName);
            t.setPriority(MIN_PRIORITY);
            t.setDaemon(true);
            return t;
        }));

        return executor.submit(() -> {
            try {
                if (model.isOnline(IGNORE_CACHE)) {
                    EventBusHolder.BUS.post(new ModelIsOnlineEvent(model));
                    model.setLastSeen(Instant.now());
                }
                Model.State state = model.getOnlineState(false);
                LOG.trace("Model online state: {} {}", model.getName(), state);
                Model.State oldState = states.getOrDefault(model, UNKNOWN);
                states.put(model, state);
                if (state != oldState) {
                    EventBusHolder.BUS.post(new ModelStateChangedEvent(model, oldState, state));
                }
            } catch (HttpException e) {
                LOG.error("Couldn't check if model {} is online. HTTP Response: {} - {}", model.getName(), e.getResponseCode(), e.getResponseMessage());
            } catch (SocketTimeoutException e) {
                LOG.error("Couldn't check if model {} is online. Request timed out", model.getName());
            } catch (InterruptedException | InterruptedIOException e) {
                Thread.currentThread().interrupt();
                if (running) {
                    LOG.error("Couldn't check if model {} is online", model.getName(), e);
                }
            } catch (Exception e) {
                LOG.error("Couldn't check if model {} is online", model.getName(), e);
            }
        });
    }

    private void suspendUntilNextIteration(List<Model> models, Duration timeCheckTook) {
        LOG.trace("Online check for {} models took {} seconds", models.size(), timeCheckTook.getSeconds());
        long sleepTime = Config.getInstance().getSettings().onlineCheckIntervalInSecs;
        if(timeCheckTook.getSeconds() < sleepTime) {
            try {
                if (running) {
                    long millis = TimeUnit.SECONDS.toMillis(sleepTime - timeCheckTook.getSeconds());
                    LOG.trace("Sleeping {}ms", millis);
                    Thread.sleep(millis);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOG.trace("Sleep interrupted");
            }
        }
    }

    public void shutdown() {
        running = false;
        for (ExecutorService executor : executors.values()) {
            executor.shutdownNow();
        }
        interrupt();
    }
}