package ctbrec.recorder.download.hls;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
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.Model;
import ctbrec.OS;
import ctbrec.Recording;
import ctbrec.io.HttpClient;
import ctbrec.recorder.FFmpeg;
import ctbrec.recorder.download.ProcessExitedUncleanException;

public class MergedFfmpegHlsDownload extends AbstractHlsDownload {

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

    protected File targetFile;
    protected transient FFmpeg ffmpeg;
    protected transient Process ffmpegProcess;
    protected transient OutputStream ffmpegStdIn;
    protected transient BlockingQueue<Future<SegmentDownload>> queue = new LinkedBlockingQueue<>();
    protected 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;
    }

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

    protected 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
    }

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