package ctbrec.recorder;

import static ctbrec.SubsequentAction.*;
import static ctbrec.event.Event.Type.*;
import static java.lang.Thread.*;
import static java.util.concurrent.TimeUnit.*;

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.time.ZoneId;
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.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
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.NoSpaceLeftEvent;
import ctbrec.event.RecordingStateChangedEvent;
import ctbrec.io.HttpClient;
import ctbrec.recorder.download.Download;
import ctbrec.recorder.postprocessing.PostProcessor;
import ctbrec.sites.Site;

public class NextGenLocalRecorder implements Recorder {

    private static final Logger LOG = LoggerFactory.getLogger(NextGenLocalRecorder.class);
    public 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 Map<Model, Recording> recordingProcesses = Collections.synchronizedMap(new HashMap<>());
    private RecordingManager recordingManager;
    private RecordingPreconditions preconditions;

    // thread pools for downloads and post-processing
    private ScheduledExecutorService downloadPool;
    private ThreadPoolScaler threadPoolScaler;
    private ExecutorService segmentDownloadPool = new ThreadPoolExecutor(0, 1000, 30L, SECONDS, new SynchronousQueue<>(), createThreadFactory("SegmentDownload", MAX_PRIORITY - 2));
    private ExecutorService downloadCompletionPool = Executors.newFixedThreadPool(1, createThreadFactory("DownloadCompletionWorker", MAX_PRIORITY - 1));
    private BlockingQueue<ScheduledFuture<Recording>> downloadFutureQueue = new LinkedBlockingQueue<>();
    private Map<ScheduledFuture<Recording>, Recording> downloadFutureRecordingMap = Collections.synchronizedMap(new HashMap<>());

    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;
        downloadPool = Executors.newScheduledThreadPool(5, createThreadFactory("Download", MAX_PRIORITY));
        threadPoolScaler = new ThreadPoolScaler((ThreadPoolExecutor) downloadPool, 5);
        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", MIN_PRIORITY));

        recording = true;
        registerEventBusListener();

        preconditions = new RecordingPreconditions(this);

        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();
                    EventBusHolder.BUS.post(new NoSpaceLeftEvent());
                }
            } catch (IOException e) {
                LOG.error("Couldn't check space left on device", e);
            }
        }, 1, 1, TimeUnit.SECONDS);
    }

    private void startCompletionHandler() {
        downloadCompletionPool.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    ScheduledFuture<Recording> future = downloadFutureQueue.take();
                    rescheduleRecordingTask(future);
                    threadPoolScaler.tick();
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    LOG.error("Error while getting recording result from download queue", e);
                } catch(Exception e) {
                    LOG.error("Error in completion handler", e);
                }
            }
        });
    }

    private void rescheduleRecordingTask(ScheduledFuture<Recording> future) throws InterruptedException{
        try {
            if (!future.isDone()) {
                downloadFutureQueue.put(future);
            } else {
                downloadFutureRecordingMap.remove(future);
                Recording rec = future.get();
                Download d = rec.getDownload();
                if (d.isRunning()) {
                    long delay = Math.max(0, Duration.between(Instant.now(), d.getRescheduleTime()).toMillis());
                    ScheduledFuture<Recording> rescheduledFuture = downloadPool.schedule(rec, delay, TimeUnit.MILLISECONDS);
                    downloadFutureQueue.add(rescheduledFuture);
                } else {
                    segmentDownloadPool.submit(() -> {
                        deleteIfEmpty(rec);
                        removeRecordingProcess(rec);
                        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) {
            fail(future, e);
        }
    }

    private void fail(ScheduledFuture<Recording> future, Exception e) {
        if (downloadFutureRecordingMap.containsKey(future)) {
            Recording rec = downloadFutureRecordingMap.remove(future);
            removeRecordingProcess(rec);
            rec.getDownload().finalizeDownload();
            LOG.error("Error while recording stream for model {}", rec.getModel(), e);
        } else {
            LOG.error("Error while recording stream", e);
        }
    }

    private void removeRecordingProcess(Recording rec) {
        recorderLock.lock();
        try {
            recordingProcesses.remove(rec.getModel());
        } finally {
            recorderLock.unlock();
        }
    }

    private void submitPostProcessingJob(Recording recording) {
        ppPool.submit(() -> {
            try {
                setRecordingStatus(recording, State.POST_PROCESSING);
                recording.getDownload().finalizeDownload();
                recording.refresh();
                recordingManager.saveRecording(recording);
                recording.postprocess();
                List<PostProcessor> postProcessors = config.getSettings().postProcessors;
                for (PostProcessor postProcessor : postProcessors) {
                    LOG.debug("Running post-processor: {}", postProcessor.getName());
                    boolean continuePP = postProcessor.postprocess(recording, recordingManager, config);
                    if (!continuePP) {
                        break;
                    }
                }
                recording.refresh();
                if (recording.getStatus() != State.DELETED) {
                    setRecordingStatus(recording, State.FINISHED);
                    recordingManager.saveRecording(recording);
                }
                LOG.info("Post-processing finished for {}", recording.getModel().getName());
            } 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 addModel(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        Optional<Model> existing = findModel(model);
        if (existing.isPresent()) {
            existing.get().setSuspended(model.isSuspended());
            existing.get().setMarkedForLaterRecording(model.isMarkedForLaterRecording());
            startRecordingProcess(existing.get());
        } else {
            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();
            }
            startRecordingProcess(model);
        }
    }

    private CompletableFuture<Void> startRecordingProcess(Model model) throws IOException {
        return CompletableFuture.runAsync(() -> {
            recorderLock.lock();
            try {
                preconditions.check(model);
                LOG.info("Starting recording for model {}", model.getName());
                Download download = createDownload(model);
                Recording rec = createRecording(download);
                setRecordingStatus(rec, State.RECORDING);
                rec.getModel().setLastRecorded(rec.getStartDate());
                recordingManager.saveRecording(rec);
                ScheduledFuture<Recording> future = downloadPool.schedule(rec, 0, TimeUnit.MILLISECONDS);
                downloadFutureQueue.add(future);
                downloadFutureRecordingMap.put(future, rec);
            } catch (RecordUntilExpiredException e) {
                LOG.info("Precondition not met to record {}: {}", model, e.getLocalizedMessage());
                executeRecordUntilSubsequentAction(model);
            } catch (PreconditionNotMetException e) {
                LOG.info("Precondition not met to record {}: {}", model, e.getLocalizedMessage());
            } catch (IOException e) {
                LOG.error("Couldn't start recording process for {}", model, e);
            } finally {
                recorderLock.unlock();
            }
        }, segmentDownloadPool);
    }

    private Download createDownload(Model model) throws IOException {
        Download download = model.createDownload();
        download.init(config, model, Instant.now(), segmentDownloadPool);
        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());
        return download;
    }

    private void executeRecordUntilSubsequentAction(Model model) {
        if (model.getRecordUntilSubsequentAction() == PAUSE) {
            model.setSuspended(true);
        } else if (model.getRecordUntilSubsequentAction() == REMOVE) {
            try {
                LOG.info("Removing {} because the recording timeframe ended at {}", model, model.getRecordUntil().atZone(ZoneId.systemDefault()));
                stopRecording(model);
            } catch (Exception e1) {
                LOG.error("Error while stopping recording", e1);
            }
        }
        // reset values, so that model can be recorded again
        model.setRecordUntil(null);
        model.setRecordUntilSubsequentAction(PAUSE);
    }

    private Recording createRecording(Download download) throws IOException {
        Model model = download.getModel();
        Recording rec = new Recording();
        rec.setId(UUID.randomUUID().toString());
        rec.setDownload(download);
        String recordingFile = download.getPath(model).replace('\\', '/');
        File absoluteFile = new File(config.getSettings().recordingsDir, recordingFile);
        rec.setAbsoluteFile(absoluteFile);
        rec.setModel(model);
        rec.setStartDate(download.getStartTime());
        rec.setSingleFile(download.isSingleFile());
        recordingProcesses.put(model, rec);
        recordingManager.add(rec);
        return rec;
    }

    private boolean deleteIfEmpty(Recording rec) {
        boolean deleted = false;
        try {
            rec.refresh();
            long sizeInByte = rec.getSizeInByte();
            if (sizeInByte <= 0) {
                LOG.info("Deleting empty recording {}", rec);
                delete(rec);
                deleted = true;
            }
            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);
            }
        } catch (IOException e) {
            LOG.error("Couldn't execute post-processing step \"delete if empty\"", e);
        }
        return deleted;
    }

    @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());
        }
    }

    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 List<Model> getModels() {
        recorderLock.lock();
        try {
            return new ArrayList<>(models);
        } finally {
            recorderLock.unlock();
        }
    }

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

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

    @Override
    public void shutdown(boolean immediately) {
        LOG.info("Shutting down");
        recording = false;

        if (!immediately) {
            stopRecordingProcesses();
            awaitDownloadsFinish();
            shutdownThreadPools();
        }
    }

    private void awaitDownloadsFinish() {
        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);
                }
            }
        }
    }

    private void shutdownThreadPools() {
        try {
            LOG.info("Shutting down download pool");
            downloadPool.shutdown();
            client.shutdown();
            downloadPool.awaitTermination(1, TimeUnit.MINUTES);
            LOG.info("Shutting down post-processing pool");
            ppPool.shutdown();
            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, InvalidKeyException, NoSuchAlgorithmException {
        recorderLock.lock();
        try {
            if (models.contains(model)) {
                int index = models.indexOf(model);
                Model m = models.get(index);
                m.setSuspended(false);
                m.setMarkedForLaterRecording(false);
                model.setSuspended(false);
                model.setMarkedForLaterRecording(false);
                config.save();
                startRecordingProcess(m);
            } else {
                LOG.warn("Couldn't resume model {}. Not found in list", model.getName());
            }
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public boolean isTracked(Model model) {
        Optional<Model> m = findModel(model);
        boolean markedForRecording = m.map(Model::isMarkedForLaterRecording).orElse(false);
        return m.isPresent() && !markedForRecording;
    }

    @Override
    public boolean isSuspended(Model model) {
        return findModel(model).map(Model::isSuspended).orElse(false);
    }

    @Override
    public boolean isMarkedForLaterRecording(Model model) {
        return findModel(model).map(Model::isMarkedForLaterRecording).orElse(false);
    }

    private Optional<Model> findModel(Model m) {
        recorderLock.lock();
        try {
            int index = models.indexOf(m);
            if (index >= 0) {
                return Optional.of(models.get(index));
            } else {
                return Optional.empty();
            }
        } finally {
            recorderLock.unlock();
        }
    }

    @Override
    public List<Model> getOnlineModels() {
        return getModels().stream().filter(m -> {
            try {
                return m.isOnline();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            } catch (Exception e) {
                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;
    }

    boolean enoughSpaceForRecording() throws IOException {
        long minimum = config.getSettings().minimumSpaceLeftInBytes;
        if (minimum == 0) { // 0 means don't check
            return getFreeSpaceBytes() > 100 * 1024 * 1024; // leave at least 100 MiB free
        } else {
            return getFreeSpaceBytes() > minimum;
        }
    }

    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, int priority) {
        return r -> {
            Thread t = new Thread(r);
            t.setName(name + " " + UUID.randomUUID().toString().substring(0, 8));
            t.setDaemon(true);
            t.setPriority(priority);
            return t;
        };
    }

    @Override
    public void rerunPostProcessing(Recording recording) throws IOException {
        recording.setPostProcessedFile(null);
        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(), segmentDownloadPool);
                other.setDownload(download);
                other.setPostProcessedFile(null);
                other.setStatus(State.WAITING);
                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());
            }
        } 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);
    }

    @Override
    public void setNote(Recording rec, String note) throws IOException {
        recordingManager.setNote(rec, note);
    }

    @Override
    public void stopRecordingAt(Model model) throws IOException, InvalidKeyException, NoSuchAlgorithmException {
        recorderLock.lock();
        try {
            int index = models.indexOf(model);
            if (index >= 0) {
                Model m = models.get(index);
                m.setRecordUntil(model.getRecordUntil());
                m.setRecordUntilSubsequentAction(model.getRecordUntilSubsequentAction());
                config.save();
            } else {
                throw new NoSuchElementException("Model " + model.getName() + " [" + model.getUrl() + "] not found in list of recorded models");
            }
        } finally {
            recorderLock.unlock();
        }

        if (Instant.now().isAfter(model.getRecordUntil())) {
            executeRecordUntilSubsequentAction(model);
        }
    }

    boolean isRecording() {
        return recording;
    }

    Map<Model, Recording> getRecordingProcesses() {
        return recordingProcesses;
    }

    @Override
    public void pause() throws InvalidKeyException, NoSuchAlgorithmException, IOException {
        LOG.info("Pausing recorder");
        try {
            recording = false;
            stopRecordingProcesses();
        } catch (Exception e) {
            recording = true;
            throw e;
        }
    }

    @Override
    public void resume() throws InvalidKeyException, NoSuchAlgorithmException, IOException {
        LOG.info("Resuming recorder");
        recording = true;
    }

    @Override
    public int getModelCount() {
        return (int) models.stream().filter(m -> !m.isMarkedForLaterRecording()).count();
    }
}
