package ctbrec.recorder.download.hls;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URL;
import java.nio.file.Files;
import java.time.Instant;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;

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

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.recorder.FFmpeg;
import ctbrec.recorder.ProgressListener;
import ctbrec.recorder.download.ProcessExitedUncleanException;
import okhttp3.Request;
import okhttp3.Response;

public class MergedFfmpegHlsDownload extends AbstractHlsDownload {

    private static final Logger LOG = LoggerFactory.getLogger(MergedFfmpegHlsDownload.class);

    private File targetFile;
    private transient FFmpeg ffmpeg;
    private transient Process ffmpegProcess;
    private transient OutputStream ffmpegStdIn;
    private transient BlockingQueue<Future<SegmentDownload>> queue = new LinkedBlockingQueue<>();
    private transient Lock ffmpegStreamLock = new ReentrantLock();

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

    @Override
    public void init(Config config, Model model, Instant startTime, ExecutorService executorService) throws IOException {
        LOG.debug("init start");
        super.init(config, model, startTime, executorService);
        String fileSuffix = config.getSettings().ffmpegFileSuffix;
        targetFile = config.getFileForRecording(model, fileSuffix, startTime);

        createTargetDirectory();
        startFfmpegProcess(targetFile);
        if (ffmpegProcess == null) {
            throw new ProcessExitedUncleanException("Couldn't spawn FFmpeg");
        }
    }

    @Override
    public AbstractHlsDownload call() throws Exception {
        super.call();
        try {
            if (!ffmpegProcess.isAlive()) {
                running = false;
                int exitValue = ffmpegProcess.exitValue();
                ffmpeg.shutdown(exitValue);
            }
        } catch (ProcessExitedUncleanException e) {
            LOG.error("FFmpeg exited unclean", e);
        }

        streamSegmentDataToFfmpeg();

        return this;
    }

    private void streamSegmentDataToFfmpeg() {
        downloadExecutor.submit(() -> {
            ffmpegStreamLock.lock();
            try {
                while (!queue.isEmpty() && !Thread.currentThread().isInterrupted()) {
                    try {
                        Future<SegmentDownload> future = queue.peek();
                        if (running && future.isDone()) {
                            queue.take();
                            SegmentDownload segmentDownload = future.get();
                            ByteArrayOutputStream downloadData = (ByteArrayOutputStream) segmentDownload.getOutputStream();
                            downloadData.writeTo(ffmpegStdIn);
                        } else {
                            // first download in queue not finished, let's continue with other stuff
                            break;
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } catch (Exception e) {
                        LOG.error("Segment download failed for model {}", model, e);
                    }
                }
            } finally {
                ffmpegStreamLock.unlock();
            }
        });
    }

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

    @Override
    public void finalizeDownload() {
        internalStop();
    }

    private void startFfmpegProcess(File target) {
        try {
            String[] cmdline = prepareCommandLine(target);
            ffmpeg = new FFmpeg.Builder()
                    .logOutput(config.getSettings().logFFmpegOutput)
                    .onStarted(p -> {
                        ffmpegProcess = p;
                        ffmpegStdIn = ffmpegProcess.getOutputStream();
                    })
                    .build();
            ffmpeg.exec(cmdline, new String[0], target.getParentFile());
        } catch (IOException | ProcessExitedUncleanException e) {
            LOG.error("Error in FFmpeg thread", e);
        }
    }

    private String[] prepareCommandLine(File target) {
        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();
        return OS.getFFmpegCommand(argsPlusFile);
    }

    @Override
    protected void execute(SegmentDownload segmentDownload) {
        queue.add(downloadExecutor.submit(segmentDownload));
    }

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

    @Override
    protected synchronized void internalStop() {
        running = false;
        for (Future<SegmentDownload> future : queue) {
            future.cancel(true);
        }
        queue.clear();

        if (ffmpegStdIn != null) {
            try {
                ffmpegStdIn.close();
            } catch (IOException e) {
                LOG.error("Couldn't terminate FFmpeg by closing stdin", e);
            }
        }

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

    @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) {
        // nothing to do
    }

    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 boolean isSingleFile() {
        return true;
    }

    @Override
    public long getSizeInByte() {
        return getTarget().length();
    }

    protected void createTargetDirectory() throws IOException {
        Files.createDirectories(targetFile.getParentFile().toPath());
    }

    @Override
    protected OutputStream getSegmentOutputStream(String prefix, String fileName) throws IOException {
        return new ByteArrayOutputStream();
    }
}
