package ctbrec.recorder;

import static ctbrec.event.Event.Type.*;

import java.io.File;
import java.io.IOException;
import java.nio.file.FileStore;
import java.nio.file.Files;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

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

import com.google.common.eventbus.Subscribe;

import ctbrec.Config;
import ctbrec.Model;
import ctbrec.Recording;
import ctbrec.Recording.State;
import ctbrec.event.Event;
import ctbrec.event.EventBusHolder;
import ctbrec.event.ModelIsOnlineEvent;
import ctbrec.event.RecordingStateChangedEvent;
import ctbrec.io.HttpClient;
import ctbrec.recorder.download.Download;
import ctbrec.sites.Site;

public class NextGenLocalRecorder implements Recorder {

    private static final Logger LOG = LoggerFactory.getLogger(NextGenLocalRecorder.class);
    private static final boolean IGNORE_CACHE = true;
    private List<Model> models = Collections.synchronizedList(new ArrayList<>());
    private Config config;
    private volatile boolean recording = true;
    private ReentrantLock recorderLock = new ReentrantLock();
    private RecorderHttpClient client = new RecorderHttpClient();
    private long lastSpaceMessage = 0;
    private Map<Model, Recording> recordingProcesses = Collections.synchronizedMap(new HashMap<>());
    private RecordingManager recordingManager;

    // thread pools for downloads and post-processing
    private BlockingQueue<Runnable> downloadQueue = new SynchronousQueue<>();
    private ThreadPoolExecutor downloadPool = new ThreadPoolExecutor(2, 100, 5, TimeUnit.MINUTES, downloadQueue, createThreadFactory("Download"));

    private ExecutorCompletionService<Recording> completionService = new ExecutorCompletionService<>(downloadPool);
    private BlockingQueue<Runnable> ppQueue = new LinkedBlockingQueue<>();
    private ThreadPoolExecutor ppPool;

    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public NextGenLocalRecorder(Config config, List<Site> sites) throws IOException {
        this.config = config;
        recordingManager = new RecordingManager(config, sites);
        config.getSettings().models.stream().forEach(m -> {
            if (m.getSite() != null) {
                if (m.getSite().isEnabled()) {
                    models.add(m);
                } else {
                    LOG.info("{} disabled -> ignoring {}", m.getSite().getName(), m.getName());
                }
            } else {
                LOG.info("Site for model {} is unknown -> ignoring", m.getName());
            }
        });

        int ppThreads = config.getSettings().postProcessingThreads;
        ppPool = new ThreadPoolExecutor(ppThreads, ppThreads, 5, TimeUnit.MINUTES, ppQueue, createThreadFactory("PP"));

        recording = true;
        registerEventBusListener();

        LOG.debug("Recorder initialized");
        LOG.info("Models to record: {}", models);
        LOG.info("Saving recordings in {}", config.getSettings().recordingsDir);

        startCompletionHandler();

        scheduler.scheduleWithFixedDelay(() -> {
            try {
                if (!recordingProcesses.isEmpty() && !enoughSpaceForRecording()) {
                    LOG.info("No space left -> Stopping all recordings");
                    stopRecordingProcesses();
                }
            } catch (IOException e) {
                LOG.error("Couldn't check space left on device", e);
            }
        }, 1, 1, TimeUnit.SECONDS);
    }

    private void startCompletionHandler() {
        Thread completionHandler = new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    Future<Recording> result = completionService.take();
                    Recording rec = result.get();
                    recorderLock.lock();
                    try {
                        recordingProcesses.remove(rec.getModel());
                    } finally {
                        recorderLock.unlock();
                    }
                    if (rec.getStatus() == State.WAITING) {
                        LOG.info("Download finished for {} -> Starting post-processing", rec.getModel().getName());
                        submitPostProcessingJob(rec);

                        // check, if we have to restart the recording
                        Model model = rec.getModel();
                        tryRestartRecording(model);
                    } else {
                        setRecordingStatus(rec, State.FAILED);
                    }
                } catch (ExecutionException | IllegalStateException e) {
                    LOG.error("Error while completing recording", e);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    LOG.error("Error while completing recording", e);
                }
            }
        });
        completionHandler.setName("CompletionHandler");
        completionHandler.setDaemon(true);
        completionHandler.start();
    }

    private void submitPostProcessingJob(Recording recording) {
        ppPool.submit(() -> {
            try {
                setRecordingStatus(recording, State.POST_PROCESSING);
                recordingManager.saveRecording(recording);
                recording.postprocess();
                setRecordingStatus(recording, State.FINISHED);
                recordingManager.saveRecording(recording);
                deleteIfTooShort(recording);
                LOG.info("Post-processing finished for {}", recording.getModel().getName());
                if (config.getSettings().removeRecordingAfterPostProcessing) {
                    recordingManager.remove(recording);
                }
            } catch (Exception e) {
                if (e instanceof InterruptedException) { // NOSONAR
                    Thread.currentThread().interrupt();
                }
                LOG.error("Error while post-processing recording {}", recording, e);
                recording.setStatus(State.FAILED);
                try {
                    recordingManager.saveRecording(recording);
                } catch (IOException e1) {
                    LOG.error("Couldn't update recording state for recording {}", recording, e1);
                }
            }
        });
    }

    private void setRecordingStatus(Recording recording, State status) {
        recording.setStatus(status);
        RecordingStateChangedEvent evt = new RecordingStateChangedEvent(recording.getDownload().getTarget(), status, recording.getModel(),
                recording.getStartDate());
        EventBusHolder.BUS.post(evt);
    }

    @Override
    public void startRecording(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        if (!models.contains(model)) {
            LOG.info("Model {} added", model);
            recorderLock.lock();
            try {
                models.add(model);
                config.getSettings().models.add(model);
                config.save();
            } catch (IOException e) {
                LOG.error("Couldn't save config", e);
            } finally {
                recorderLock.unlock();
            }

            // try to start the recording immediately
            try {
                if (model.isOnline()) {
                    startRecordingProcess(model);
                }
            } catch (ExecutionException e) {
                // noop
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    private void startRecordingProcess(Model model) throws IOException {
        recorderLock.lock();
        try {
            if (!recording) {
                // recorder is not in recording mode
                return;
            }

            if (model.isSuspended()) {
                LOG.info("Recording for model {} is suspended.", model);
                return;
            }

            if (recordingProcesses.containsKey(model)) {
                LOG.error("A recording for model {} is already running", model);
                return;
            }

            if (!models.contains(model)) {
                LOG.info("Model {} has been removed. Restarting of recording cancelled.", model);
                return;
            }

            if (!enoughSpaceForRecording()) {
                long now = System.currentTimeMillis();
                if ((now - lastSpaceMessage) > TimeUnit.MINUTES.toMillis(1)) {
                    LOG.info("Not enough space for recording, not starting recording for {}", model);
                    lastSpaceMessage = now;
                }
                return;
            }

            if (!downloadSlotAvailable()) {
                long now = System.currentTimeMillis();
                if ((now - lastSpaceMessage) > TimeUnit.MINUTES.toMillis(1)) {
                    LOG.info("The number of downloads is maxed out");
                }
                // check, if we can stop a recording for a model with lower priority
                Optional<Recording> lowerPrioRecordingProcess = recordingProcessWithLowerPrio(model.getPriority());
                if (lowerPrioRecordingProcess.isPresent()) {
                    Download download = lowerPrioRecordingProcess.get().getDownload();
                    Model lowerPrioModel = download.getModel();
                    LOG.info("Stopping recording for {}. Prio {} < {}", lowerPrioModel.getName(), lowerPrioModel.getPriority(), model.getPriority());
                    stopRecordingProcess(lowerPrioModel);
                } else {
                    if ((now - lastSpaceMessage) > TimeUnit.MINUTES.toMillis(1)) {
                        LOG.info("Other models have higher prio, not starting recording for {}", model.getName());
                    }
                    return;
                }
            }

            LOG.info("Starting recording for model {}", model.getName());
            Download download = model.createDownload();
            download.init(config, model, Instant.now());
            Objects.requireNonNull(download.getStartTime(),
                    "At this point the download should have set a startTime. Make sure to set a startTime in " + download.getClass() + ".init()");
            LOG.debug("Downloading with {}", download.getClass().getSimpleName());

            Recording rec = new Recording();
            rec.setDownload(download);
            rec.setPath(download.getPath(model).replaceAll("\\\\", "/"));
            rec.setModel(model);
            rec.setStartDate(download.getStartTime());
            rec.setSingleFile(download.isSingleFile());
            recordingProcesses.put(model, rec);
            recordingManager.add(rec);
            completionService.submit(() -> {
                try {
                    setRecordingStatus(rec, State.RECORDING);
                    model.setLastRecorded(rec.getStartDate());
                    recordingManager.saveRecording(rec);
                    download.start();
                } catch (Exception e) {
                    LOG.error("Download for {} failed. Download state: {}", model.getName(), rec.getStatus(), e);
                }
                boolean deleted = deleteIfEmpty(rec);
                setRecordingStatus(rec, deleted ? State.DELETED : State.WAITING);
                if (!deleted) {
                    // only save the status, if the recording has not been deleted, otherwise we recreate the metadata file
                    recordingManager.saveRecording(rec);
                }
                return rec;
            });
        } finally {
            recorderLock.unlock();
        }
    }

    private Optional<Recording> recordingProcessWithLowerPrio(int priority) {
        Model lowest = null;
        for (Model m : recordingProcesses.keySet()) {
            if (lowest == null || m.getPriority() < lowest.getPriority()) {
                lowest = m;
            }
        }
        if (lowest != null && lowest.getPriority() < priority) {
            return Optional.of(recordingProcesses.get(lowest));
        } else {
            return Optional.empty();
        }
    }

    private boolean deleteIfEmpty(Recording rec) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        rec.refresh();
        long sizeInByte = rec.getSizeInByte();
        if (sizeInByte <= 0) {
            LOG.info("Deleting empty recording {}", rec);
            delete(rec);
            return true;
        } else {
            return false;
        }
    }

    private boolean deleteIfTooShort(Recording rec) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        Duration minimumLengthInSeconds = Duration.ofSeconds(Config.getInstance().getSettings().minimumLengthInSeconds);
        if (minimumLengthInSeconds.getSeconds() <= 0) {
            return false;
        }

        Duration recordingLength = rec.getLength();
        if (recordingLength.compareTo(minimumLengthInSeconds) < 0) {
            LOG.info("Deleting too short recording {} [{} < {}]", rec, recordingLength, minimumLengthInSeconds);
            delete(rec);
            return true;
        }

        return false;
    }

    @Override
    public void stopRecording(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recorderLock.lock();
        try {
            if (models.contains(model)) {
                models.remove(model);
                config.getSettings().models.remove(model);
                LOG.info("Model {} removed", model);
                config.save();
            } else {
                throw new NoSuchElementException("Model " + model.getName() + " [" + model.getUrl() + "] not found in list of recorded models");
            }

            if (recordingProcesses.containsKey(model)) {
                Recording rec = recordingProcesses.get(model);
                rec.getDownload().stop();
            }
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public void switchStreamSource(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        if (models.contains(model)) {
            int index = models.indexOf(model);
            models.get(index).setStreamUrlIndex(model.getStreamUrlIndex());
            config.save();
            LOG.debug("Switching stream source to index {} for model {}", model.getStreamUrlIndex(), model.getName());
            recorderLock.lock();
            try {
                Recording rec = recordingProcesses.get(model);
                if (rec != null) {
                    stopRecordingProcess(model);
                }
            } finally {
                recorderLock.unlock();
            }
            tryRestartRecording(model);
        } else {
            LOG.warn("Couldn't switch stream source for model {}. Not found in list", model.getName());
        }
    }

    private void stopRecordingProcess(Model model) {
        recorderLock.lock();
        try {
            LOG.debug("Stopping recording for {}", model);
            Recording rec = recordingProcesses.get(model);
            LOG.debug("Stopping download for {}", model);
            rec.getDownload().stop();
        } finally {
            recorderLock.unlock();
        }
    }

    private void stopRecordingProcesses() {
        recorderLock.lock();
        try {
            for (Recording rec : recordingProcesses.values()) {
                rec.getDownload().stop();
            }
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public boolean isTracked(Model model) {
        recorderLock.lock();
        try {
            return models.contains(model);
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public List<Model> getModels() {
        recorderLock.lock();
        try {
            return new ArrayList<>(models);
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public List<Recording> getRecordings() throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        return recordingManager.getAll();
    }

    @Override
    public void delete(Recording recording) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recordingManager.delete(recording);
    }

    @Override
    public void shutdown() {
        // TODO add a config flag for waitign or stopping immediately
        LOG.info("Shutting down");
        recording = false;

        LOG.debug("Stopping all recording processes");
        recorderLock.lock();
        try {
            // make a copy to avoid ConcurrentModificationException
            List<Recording> toStop = new ArrayList<>(recordingProcesses.values());
            for (Recording rec : toStop) {
                Optional.ofNullable(rec.getDownload()).ifPresent(Download::stop);
            }
        } finally {
            recorderLock.unlock();
        }

        // wait for post-processing to finish
        LOG.info("Waiting for downloads to finish");
        for (int i = 0; i < 60; i++) {
            if (!recordingProcesses.isEmpty()) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    LOG.error("Error while waiting for downloads to finish", e);
                }
            }
        }

        // shutdown threadpools
        try {
            LOG.info("Shutting down pools");
            downloadPool.shutdown();
            ppPool.shutdown();
            client.shutdown();
            downloadPool.awaitTermination(1, TimeUnit.MINUTES);
            int minutesToWait = 10;
            LOG.info("Waiting {} minutes (max) for post-processing to finish", minutesToWait);
            ppPool.awaitTermination(minutesToWait, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.error("Error while waiting for pools to finish", e);
        }
    }

    @Override
    public void suspendRecording(Model model) {
        recorderLock.lock();
        try {
            if (models.contains(model)) {
                int index = models.indexOf(model);
                models.get(index).setSuspended(true);
                model.setSuspended(true);
                config.save();
            } else {
                LOG.warn("Couldn't suspend model {}. Not found in list", model.getName());
                return;
            }

            Recording rec = recordingProcesses.get(model);
            Optional.ofNullable(rec).map(Recording::getDownload).ifPresent(Download::stop);
        } catch (IOException e) {
            LOG.error("Couldn't save config", e);
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public void resumeRecording(Model model) throws IOException {
        recorderLock.lock();
        try {
            if (models.contains(model)) {
                int index = models.indexOf(model);
                Model m = models.get(index);
                m.setSuspended(false);
                if (m.isOnline()) {
                    startRecordingProcess(m);
                }
                model.setSuspended(false);
                config.save();
            } else {
                LOG.warn("Couldn't resume model {}. Not found in list", model.getName());
            }
        } catch (ExecutionException e) {
            LOG.error("Couldn't check, if model {} is online", model.getName());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.error("Couldn't check, if model {} is online", model.getName());
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public boolean isSuspended(Model model) {
        recorderLock.lock();
        try {
            int index = models.indexOf(model);
            if (index >= 0) {
                Model m = models.get(index);
                return m.isSuspended();
            } else {
                return false;
            }
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public List<Model> getOnlineModels() {
        return getModels().stream().filter(m -> {
            try {
                return m.isOnline();
            } catch (IOException | ExecutionException e) {
                return false;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }).collect(Collectors.toList());
    }

    @Override
    public HttpClient getHttpClient() {
        return client;
    }

    @Override
    public long getTotalSpaceBytes() throws IOException {
        return getRecordingsFileStore().getTotalSpace();
    }

    @Override
    public long getFreeSpaceBytes() throws IOException {
        return getRecordingsFileStore().getUsableSpace();
    }

    private FileStore getRecordingsFileStore() throws IOException {
        File recordingsDir = new File(config.getSettings().recordingsDir);
        if (!recordingsDir.exists()) {
            Files.createDirectories(recordingsDir.toPath());
        }
        FileStore store = Files.getFileStore(recordingsDir.toPath());
        return store;
    }

    private boolean enoughSpaceForRecording() throws IOException {
        long minimum = config.getSettings().minimumSpaceLeftInBytes;
        if (minimum == 0) { // 0 means don't check
            return true;
        } else {
            return getFreeSpaceBytes() > minimum;
        }
    }

    private boolean downloadSlotAvailable() {
        int concurrentRecordings = Config.getInstance().getSettings().concurrentRecordings;
        return concurrentRecordings == 0 || concurrentRecordings > 0 && recordingProcesses.size() < concurrentRecordings;
    }

    private void tryRestartRecording(Model model) {
        if (!recording) {
            // recorder is not in recording state
            return;
        }

        try {
            boolean modelInRecordingList = isTracked(model);
            boolean online = model.isOnline(IGNORE_CACHE);
            if (modelInRecordingList && online) {
                LOG.info("Restarting recording for model {}", model);
                startRecordingProcess(model);
            }
        } catch (Exception e) {
            LOG.error("Couldn't restart recording for model {}", model);
        }
    }

    private void registerEventBusListener() {
        EventBusHolder.BUS.register(new Object() {
            @Subscribe
            public void modelEvent(Event e) {
                recorderLock.lock();
                try {
                    if (e.getType() == MODEL_ONLINE) {
                        ModelIsOnlineEvent evt = (ModelIsOnlineEvent) e;
                        Model model = evt.getModel();
                        if (!isSuspended(model) && !recordingProcesses.containsKey(model)) {
                            startRecordingProcess(model);
                        }
                    }
                } catch (Exception e1) {
                    LOG.error("Error while handling model state changed event {}", e, e1);
                } finally {
                    recorderLock.unlock();
                }
            }
        });
    }

    private ThreadFactory createThreadFactory(String name) {
        return r -> {
            Thread t = new Thread(r);
            t.setName(name + " " + UUID.randomUUID().toString().substring(0, 8));
            t.setDaemon(true);
            return t;
        };
    }

    @Override
    public void rerunPostProcessing(Recording recording) {
        List<Recording> recordings = recordingManager.getAll();
        for (Recording other : recordings) {
            if(other.equals(recording)) {
                Download download = other.getModel().createDownload();
                download.init(Config.getInstance(), other.getModel(), other.getStartDate());
                other.setDownload(download);
                submitPostProcessingJob(other);
                return;
            }
        }
        LOG.error("Recording {} not found. Can't rerun post-processing", recording);
    }

    @Override
    public void priorityChanged(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recorderLock.lock();
        try {
            if (models.contains(model)) {
                int index = models.indexOf(model);
                models.get(index).setPriority(model.getPriority());
                config.save();
            } else {
                LOG.warn("Couldn't change priority for model {}. Not found in list", model.getName());
                return;
            }
        } catch (IOException e) {
            LOG.error("Couldn't save config", e);
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public void pin(Recording recording) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recordingManager.pin(recording);
    }

    @Override
    public void unpin(Recording recording) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recordingManager.unpin(recording);
    }
}
