package ctbrec.recorder.download.hls;

import static ctbrec.io.HttpConstants.*;

import java.io.EOFException;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URL;
import java.nio.file.Files;
import java.time.Duration;
import java.time.Instant;
import java.time.ZonedDateTime;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

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

import com.iheartradio.m3u8.ParseException;
import com.iheartradio.m3u8.PlaylistException;

import ctbrec.Config;
import ctbrec.Hmac;
import ctbrec.Model;
import ctbrec.OS;
import ctbrec.Recording;
import ctbrec.io.HttpClient;
import ctbrec.io.HttpException;
import ctbrec.io.StreamRedirectThread;
import ctbrec.recorder.ProgressListener;
import ctbrec.recorder.download.ProcessExitedUncleanException;
import okhttp3.Request;
import okhttp3.Response;

public class MergedFfmpegHlsDownload extends AbstractHlsDownload {

    private static final long serialVersionUID = 1L;
    private static final Logger LOG = LoggerFactory.getLogger(MergedFfmpegHlsDownload.class);
    private static final boolean IGNORE_CACHE = true;
    private ZonedDateTime splitRecStartTime;
    private File targetFile;
    private transient Config config;
    private transient Process ffmpeg;
    private transient OutputStream ffmpegStdIn;
    protected transient Thread ffmpegThread;
    private transient Object ffmpegStartMonitor = new Object();

    public MergedFfmpegHlsDownload(HttpClient client) {
        super(client);
    }

    @Override
    public void init(Config config, Model model, Instant startTime) {
        super.startTime = startTime;
        this.config = config;
        this.model = model;
        String fileSuffix = config.getSettings().ffmpegFileSuffix;
        targetFile = config.getFileForRecording(model, fileSuffix, startTime);
    }

    @Override
    public File getTarget() {
        return targetFile;
    }

    @Override
    public void start() throws IOException {
        try {
            if (!model.isOnline(IGNORE_CACHE)) {
                throw new IOException(model.getName() + "'s room is not public");
            }

            running = true;
            super.startTime = Instant.now();
            splitRecStartTime = ZonedDateTime.now();

            String segments = getSegmentPlaylistUrl(model);

            Files.createDirectories(targetFile.getParentFile().toPath());
            startFfmpegProcess(targetFile);
            synchronized (ffmpegStartMonitor) {
                int tries = 0;
                while (ffmpeg == null && tries++ < 15) {
                    LOG.debug("Waiting for FFmpeg to spawn to record {}", model.getName());
                    ffmpegStartMonitor.wait(1000);
                }
            }

            if (ffmpeg == null) {
                throw new ProcessExitedUncleanException("Couldn't spawn FFmpeg");
            } else {
                LOG.debug("Starting to download segments");
                downloadSegments(segments, true);
                ffmpegThread.join();
            }
        } catch (ParseException e) {
            throw new IOException("Couldn't parse stream information", e);
        } catch (PlaylistException e) {
            throw new IOException("Couldn't parse HLS playlist", e);
        } catch (EOFException e) {
            // end of playlist reached
            LOG.debug("Reached end of playlist for model {}", model);
        } catch (Exception e) {
            throw new IOException("Exception while downloading segments", e);
        } finally {
            internalStop();
            downloadThreadPool.shutdown();
            try {
                LOG.debug("Waiting for last segments for {}", model);
                downloadThreadPool.awaitTermination(60, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            running = false;
            LOG.debug("Download for {} terminated", model);
        }
    }

    private void startFfmpegProcess(File target) {
        ffmpegThread = new Thread(() -> {
            try {
                String[] args = config.getSettings().ffmpegMergedDownloadArgs.split(" ");
                String[] argsPlusFile = new String[args.length + 3];
                int i = 0;
                argsPlusFile[i++] = "-i";
                argsPlusFile[i++] = "-";
                System.arraycopy(args, 0, argsPlusFile, i, args.length);
                argsPlusFile[argsPlusFile.length - 1] = target.getAbsolutePath();
                String[] cmdline = OS.getFFmpegCommand(argsPlusFile);

                LOG.debug("Command line: {}", Arrays.toString(cmdline));
                ffmpeg = Runtime.getRuntime().exec(cmdline, new String[0], target.getParentFile());
                synchronized (ffmpegStartMonitor) {
                    ffmpegStartMonitor.notifyAll();
                }
                ffmpegStdIn = ffmpeg.getOutputStream();
                int exitCode = 1;
                File ffmpegLog = File.createTempFile(target.getName(), ".log");
                try (FileOutputStream mergeLogStream = new FileOutputStream(ffmpegLog)) {
                    Thread stdout = new Thread(new StreamRedirectThread(ffmpeg.getInputStream(), mergeLogStream));
                    Thread stderr = new Thread(new StreamRedirectThread(ffmpeg.getErrorStream(), mergeLogStream));
                    stdout.start();
                    stderr.start();
                    exitCode = ffmpeg.waitFor();
                    LOG.debug("FFmpeg exited with code {}", exitCode);
                    stdout.join();
                    stderr.join();
                    mergeLogStream.flush();
                }
                if (exitCode != 1) {
                    if (ffmpegLog.exists()) {
                        Files.delete(ffmpegLog.toPath());
                    }
                } else {
                    if (running) {
                        LOG.info("FFmpeg exit code was {}. Logfile: {}", exitCode, ffmpegLog.getAbsolutePath());
                        throw new ProcessExitedUncleanException("FFmpeg exit code was " + exitCode);
                    }
                }
            } catch (IOException | ProcessExitedUncleanException e) {
                LOG.error("Error in FFMpeg thread", e);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                if (running) {
                    LOG.info("Interrupted while waiting for ffmpeg", e);
                }
            }
        });
        ffmpegThread.setName("FFmpeg");
        ffmpegThread.start();
    }

    protected void downloadSegments(String segmentPlaylistUri, boolean livestreamDownload) throws IOException, ParseException, PlaylistException {
        int lastSegment = 0;
        int nextSegment = 0;
        while (running) {
            try {
                SegmentPlaylist lsp = getNextSegments(segmentPlaylistUri);
                emptyPlaylistCheck(lsp);

                // download new segments
                long downloadStart = System.currentTimeMillis();
                if (livestreamDownload) {
                    downloadNewSegments(lsp, nextSegment);
                } else {
                    downloadRecording(lsp);
                }
                long downloadTookMillis = System.currentTimeMillis() - downloadStart;

                // download segments, which might have been skipped
                if (nextSegment > 0 && lsp.seq > nextSegment) {
                    LOG.warn("Missed segments {} < {} in download for {}. Download took {}ms. Playlist is {}sec", nextSegment, lsp.seq, lsp.url,
                            downloadTookMillis, lsp.totalDuration);
                }

                if (livestreamDownload) {
                    // split up the recording, if configured
                    boolean split = splitRecording();
                    if (split) {
                        break;
                    }

                    // wait some time until requesting the segment playlist again to not hammer the server
                    waitForNewSegments(lsp, lastSegment, downloadTookMillis);

                    lastSegment = lsp.seq;
                    nextSegment = lastSegment + lsp.segments.size();
                } else {
                    break;
                }
            } catch (HttpException e) {
                if (e.getResponseCode() == 404) {
                    LOG.debug("Playlist not found (404). Model {} probably went offline", model);
                } else if (e.getResponseCode() == 403) {
                    LOG.debug("Playlist access forbidden (403). Model {} probably went private or offline", model);
                } else {
                    LOG.info("Unexpected error while downloading {}", model, e);
                }
                running = false;
            } catch (Exception e) {
                LOG.info("Unexpected error while downloading {}", model, e);
                running = false;
            }
        }
        ffmpegThread.interrupt();
    }

    private void downloadRecording(SegmentPlaylist lsp) throws IOException {
        for (String segment : lsp.segments) {
            URL segmentUrl = new URL(segment);
            SegmentDownload segmentDownload = new SegmentDownload(lsp, segmentUrl, client);
            byte[] segmentData = segmentDownload.call();
            writeSegment(segmentData);
        }
    }

    private void downloadNewSegments(SegmentPlaylist lsp, int nextSegment) throws ExecutionException, IOException {
        int skip = nextSegment - lsp.seq;

        // add segments to download threadpool
        Queue<Future<byte[]>> downloads = new LinkedList<>();
        if (downloadQueue.remainingCapacity() == 0) {
            LOG.warn("Download to slow for this stream. Download queue is full. Skipping segment");
        } else {
            for (String segment : lsp.segments) {
                if (!running) {
                    break;
                }
                if (skip > 0) {
                    skip--;
                } else {
                    URL segmentUrl = new URL(segment);
                    Future<byte[]> download = downloadThreadPool.submit(new SegmentDownload(lsp, segmentUrl, client));
                    downloads.add(download);
                }
            }
        }

        writeFinishedSegments(downloads);
    }

    private void writeFinishedSegments(Queue<Future<byte[]>> downloads) throws ExecutionException, IOException {
        for (Future<byte[]> downloadFuture : downloads) {
            try {
                byte[] segmentData = downloadFuture.get();
                writeSegment(segmentData);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOG.error("Error while downloading segment", e);
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof MissingSegmentException) {
                    if (model != null && !isModelOnline()) {
                        LOG.debug("Error while downloading segment, because model {} is offline. Stopping now", model.getName());
                        running = false;
                    } else {
                        LOG.debug("Segment not available, but model {} still online. Going on", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
                    }
                } else if (cause instanceof HttpException) {
                    HttpException he = (HttpException) cause;
                    if (model != null && !isModelOnline()) {
                        LOG.debug("Error {} while downloading segment, because model {} is offline. Stopping now", he.getResponseCode(), model.getName());
                        running = false;
                    } else {
                        if (he.getResponseCode() == 404) {
                            LOG.info("Playlist for {} not found [HTTP 404]. Stopping now", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
                            running = false;
                        } else if (he.getResponseCode() == 403) {
                            LOG.info("Playlist for {} not accessible [HTTP 403]. Stopping now", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
                            running = false;
                        } else {
                            throw he;
                        }
                    }
                } else {
                    throw e;
                }
            }
        }
    }

    protected void writeSegment(byte[] segmentData, int offset, int length) throws IOException {
        if (running) {
            if (ffmpegStdIn != null) {
                ffmpegStdIn.write(segmentData, offset, length);
            } else {
                LOG.error("FFmpeg stdin stream is null - skipping writing of segment");
            }
        }
    }

    private void writeSegment(byte[] segmentData) throws IOException {
        writeSegment(segmentData, 0, segmentData.length);
    }

    protected boolean splitRecording() {
        if (config.getSettings().splitRecordings > 0) {
            Duration recordingDuration = Duration.between(splitRecStartTime, ZonedDateTime.now());
            long seconds = recordingDuration.getSeconds();
            if (seconds >= config.getSettings().splitRecordings) {
                internalStop();
                return true;
            }
        }
        return false;
    }

    private void waitForNewSegments(SegmentPlaylist lsp, int lastSegment, long downloadTookMillis) {
        try {
            long wait = 0;
            if (lastSegment == lsp.seq) {
                int timeLeftMillis = (int) (lsp.totalDuration * 1000 - downloadTookMillis);
                if (timeLeftMillis < 3000) { // we have less than 3 seconds to get the new playlist and start downloading it
                    wait = 1;
                } else {
                    // wait a second to be nice to the server (don't hammer it with requests)
                    // 1 second seems to be a good compromise. every other calculation resulted in more missing segments
                    wait = 1000;
                }
                LOG.trace("Playlist didn't change... waiting for {}ms", wait);
            } else {
                // playlist did change -> wait for at least last segment duration
                wait = 1;
                LOG.trace("Playlist changed... waiting for {}ms", wait);
            }
            Thread.sleep(wait);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            if (running) {
                LOG.error("Couldn't sleep between segment downloads. This might mess up the download!");
            }
        }
    }

    @Override
    public void stop() {
        if (running) {
            internalStop();
        }
    }

    @Override
    synchronized void internalStop() {
        running = false;
        if (ffmpegStdIn != null) {
            try {
                downloadQueue.clear();
                ffmpegStdIn.close();
            } catch (IOException e) {
                LOG.error("Couldn't terminate FFmpeg by closing stdin", e);
            }
        }

        if (ffmpeg != null) {
            try {
                boolean waitFor = ffmpeg.waitFor(5, TimeUnit.MINUTES);
                if (!waitFor && ffmpeg.isAlive()) {
                    LOG.info("FFmpeg didn't terminate. Destroying the process with force!");
                    ffmpeg.destroyForcibly();
                    ffmpeg = null;
                }
            } catch (InterruptedException e) {
                LOG.error("Interrupted while waiting for FFmpeg to terminate");
                Thread.currentThread().interrupt();
            }
        }
    }

    private class SegmentDownload implements Callable<byte[]> {
        private URL url;
        private HttpClient client;
        private SegmentPlaylist lsp;

        public SegmentDownload(SegmentPlaylist lsp, URL url, HttpClient client) {
            this.lsp = lsp;
            this.url = url;
            this.client = client;
        }

        @Override
        public byte[] call() throws IOException {
            LOG.trace("Downloading segment {}", url.getFile());
            int maxTries = 3;
            for (int i = 1; i <= maxTries && running; i++) {
                Request request = new Request.Builder().url(url)
                        .header(USER_AGENT, Config.getInstance().getSettings().httpUserAgent)
                        .header(CONNECTION, KEEP_ALIVE)
                        .build();
                try (Response response = client.execute(request)) {
                    if (response.isSuccessful()) {
                        byte[] segment = response.body().bytes();
                        if (lsp.encrypted) {
                            segment = new Crypto(lsp.encryptionKeyUrl, client).decrypt(segment);
                        }
                        return segment;
                    } else {
                        throw new HttpException(response.code(), response.message());
                    }
                } catch (Exception e) {
                    if (i == maxTries) {
                        LOG.error("Error while downloading segment. Segment {} finally failed", url.getFile());
                    } else {
                        LOG.trace("Error while downloading segment {} on try {}", url.getFile(), i, e);
                    }
                    if (model != null && !isModelOnline()) {
                        break;
                    }
                }
            }
            throw new MissingSegmentException("Unable to download segment " + url.getFile() + " after " + maxTries + " tries");
        }
    }

    public boolean isModelOnline() {
        try {
            return model.isOnline(IGNORE_CACHE);
        } catch (IOException | ExecutionException e) {
            return false;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }

    @Override
    public String getPath(Model model) {
        String absolutePath = targetFile.getAbsolutePath();
        String recordingsDir = Config.getInstance().getSettings().recordingsDir;
        String relativePath = absolutePath.replaceFirst(Pattern.quote(recordingsDir), "");
        return relativePath;
    }

    @Override
    public void postprocess(Recording recording) {
        Thread.currentThread().setName("PP " + model.getName());
        try {
            runPostProcessingScript(recording);
        } catch (Exception e) {
            throw new PostProcessingException(e);
        }
    }

    public void downloadFinishedRecording(String segmentPlaylistUri, File target, ProgressListener progressListener, long sizeInBytes) throws Exception {
        running = true;
        if (Config.getInstance().getSettings().requireAuthentication) {
            URL u = new URL(segmentPlaylistUri);
            String path = u.getPath();
            byte[] key = Config.getInstance().getSettings().key;
            if (!Config.getInstance().getContextPath().isEmpty()) {
                path = path.substring(Config.getInstance().getContextPath().length());
            }
            String hmac = Hmac.calculate(path, key);
            segmentPlaylistUri = segmentPlaylistUri + "?hmac=" + hmac;
        }

        startFfmpegProcess(target);
        for (int i = 0; i < 10 && ffmpegStdIn == null; i++) {
            Thread.sleep(100);
        }

        SegmentPlaylist segmentPlaylist = getNextSegments(segmentPlaylistUri);
        long loadedBytes = 0;
        for (String segmentUrl : segmentPlaylist.segments) {
            loadedBytes += downloadFile(segmentUrl, loadedBytes, sizeInBytes, progressListener);
            int progress = (int) (loadedBytes / (double) sizeInBytes * 100);
            progressListener.update(progress);
        }

        internalStop();
    }

    private long downloadFile(String fileUri, long loadedBytes, long totalBytes, ProgressListener progressListener) throws IOException {
        long fileLoadedBytes = 0;
        Request request = new Request.Builder().url(fileUri).addHeader("connection", "keep-alive").build();
        try (Response response = client.execute(request)) {
            if (response.isSuccessful()) {
                InputStream in = response.body().byteStream();
                byte[] b = new byte[1024 * 100];
                int length = -1;
                while ((length = in.read(b)) >= 0) {
                    ffmpegStdIn.write(b, 0, length);
                    fileLoadedBytes += length;
                    int progress = (int) ((loadedBytes + fileLoadedBytes) / (double) totalBytes * 100);
                    progressListener.update(progress);
                }
            } else {
                throw new HttpException(response.code(), response.message());
            }
        }
        return fileLoadedBytes;
    }

    @Override
    public Duration getLength() {
        return Duration.between(getStartTime(), Instant.now());
    }

    @Override
    public boolean isSingleFile() {
        return true;
    }
}
