package ctbrec.recorder.download.hls;

import static ctbrec.io.HttpConstants.*;
import static ctbrec.recorder.download.StreamSource.*;
import static java.nio.charset.StandardCharsets.*;
import static java.nio.file.StandardOpenOption.*;

import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.nio.file.Files;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import javax.xml.bind.JAXBException;

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

import com.iheartradio.m3u8.Encoding;
import com.iheartradio.m3u8.Format;
import com.iheartradio.m3u8.ParseException;
import com.iheartradio.m3u8.ParsingMode;
import com.iheartradio.m3u8.PlaylistException;
import com.iheartradio.m3u8.PlaylistParser;
import com.iheartradio.m3u8.data.EncryptionData;
import com.iheartradio.m3u8.data.MediaPlaylist;
import com.iheartradio.m3u8.data.Playlist;
import com.iheartradio.m3u8.data.TrackData;

import ctbrec.Config;
import ctbrec.Model;
import ctbrec.Model.State;
import ctbrec.StringUtil;
import ctbrec.io.BandwidthMeter;
import ctbrec.io.HttpClient;
import ctbrec.io.HttpConstants;
import ctbrec.io.HttpException;
import ctbrec.recorder.PlaylistGenerator.InvalidPlaylistException;
import ctbrec.recorder.download.AbstractDownload;
import ctbrec.recorder.download.HttpHeaderFactory;
import ctbrec.recorder.download.StreamSource;
import ctbrec.sites.Site;
import okhttp3.Request;
import okhttp3.Request.Builder;
import okhttp3.Response;

public abstract class AbstractHlsDownload extends AbstractDownload {

    private static final transient Logger LOG = LoggerFactory.getLogger(AbstractHlsDownload.class);
    private static final int TEN_SECONDS = 10_000;
    private static final boolean DEBUG_HLS = Objects.equals(System.getenv("CTBREC_DEBUG_HLS"), "true");

    private transient NumberFormat nf = new DecimalFormat("000000");
    private transient int playlistEmptyCount = 0;
    private transient int segmentCounter = 1;

    protected transient HttpClient client;
    protected transient volatile boolean running = true;

    protected transient int lastSegmentNumber = 0;
    protected transient int nextSegmentNumber = 0;
    protected transient String segmentPlaylistUrl;

    private transient Instant beforeLastPlaylistRequest= Instant.EPOCH;
    private transient int consecutivePlaylistTimeouts = 0;
    private transient int consecutivePlaylistErrors = 0;

    private transient List<RecordingEvent> recordingEvents = new LinkedList<>();

    protected AbstractHlsDownload(HttpClient client) {
        this.client = client;
    }

    protected abstract OutputStream getSegmentOutputStream(String prefix, String fileName) throws IOException;
    protected void segmentDownloadFinished(SegmentDownload segmentDownload) {}
    protected abstract void internalStop();

    @Override
    public AbstractHlsDownload call() throws Exception {
        try {
            if (segmentPlaylistUrl == null) {
                segmentPlaylistUrl = getSegmentPlaylistUrl(model);
            }

            beforeLastPlaylistRequest = Instant.now();
            SegmentPlaylist segmentPlaylist = getNextSegments(segmentPlaylistUrl);
            emptyPlaylistCheck(segmentPlaylist);
            handleMissedSegments(segmentPlaylist, nextSegmentNumber);
            enqueueNewSegments(segmentPlaylist, nextSegmentNumber);
            splitRecordingIfNecessary();
            calculateRescheduleTime();

            // this if check makes sure, that we don't decrease nextSegment. for some reason
            // streamate playlists sometimes jump back. e.g. max sequence = 79 -> 80 -> 79
            lastSegmentNumber = segmentPlaylist.seq;
            if (lastSegmentNumber + segmentPlaylist.segments.size() > nextSegmentNumber) {
                nextSegmentNumber = lastSegmentNumber + segmentPlaylist.segments.size();
            }

            while (recordingEvents.size() > 30) {
                recordingEvents.remove(0);
            }
        } catch (ParseException e) {
            LOG.error("Couldn't parse HLS playlist for model {}\n{}", model, e.getInput(), e);
            stop();
        } catch (PlaylistException e) {
            LOG.error("Couldn't parse HLS playlist for model {}", model, e);
            stop();
        } catch (PlaylistTimeoutException e) {
            rescheduleTime = beforeLastPlaylistRequest; // try again as fast as possible
        } catch (EOFException e) {
            // end of playlist reached
            LOG.debug("Reached end of playlist for model {}", model);
            stop();
        } catch (HttpException e) {
            handleHttpException(e);
        } catch (Exception e) {
            LOG.error("Couldn't download segment for model {}", model, e);
            stop();
        }
        return this;
    }

    protected void execute(SegmentDownload segmentDownload) {
        CompletableFuture.supplyAsync(() -> downloadExecutor.submit(segmentDownload), downloadExecutor)
        .whenComplete((result, executor) -> {
            try {
                segmentDownloadFinished(result.get());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOG.error("Error in segmentDownloadFinished", e);
            } catch (ExecutionException e) {
                LOG.error("Error in segmentDownloadFinished", e);
            }
        });
    }

    private void handleHttpException(HttpException e) throws IOException {
        if (e.getResponseCode() == 404) {
            checkIfModelIsStillOnline("Playlist not found (404). Model {} probably went offline. Model state: {}");
        } else if (e.getResponseCode() == 403) {
            checkIfModelIsStillOnline("Playlist access forbidden (403). Model {} probably went private or offline. Model state: {}");
        } else {
            if (consecutivePlaylistErrors >= 3) {
                LOG.info("Playlist couldn't not be downloaded for model {} {} times. Stopping recording", model, (consecutivePlaylistErrors + 1), e);
                stop();
            } else {
                LOG.info("Playlist couldn't not be downloaded for model {} {} times: {}", model, (consecutivePlaylistErrors + 1), e.getLocalizedMessage());
            }
        }
    }

    protected void checkIfModelIsStillOnline(String errorMsg) throws IOException {
        ctbrec.Model.State modelState;
        try {
            modelState = model.getOnlineState(false);
            if (modelState != State.ONLINE) {
                stop();
            }
        } catch (ExecutionException e1) {
            modelState = State.UNKNOWN;
        }
        LOG.info(errorMsg, model, modelState);
        waitSomeTime(TEN_SECONDS);
    }

    protected String getSegmentPlaylistUrl(Model model) throws IOException, ExecutionException, ParseException, PlaylistException, JAXBException {
        LOG.debug("{} stream idx: {}", model.getName(), model.getStreamUrlIndex());
        List<StreamSource> streamSources = model.getStreamSources();
        Collections.sort(streamSources);
        for (StreamSource streamSource : streamSources) {
            LOG.debug("{} src {}", model.getName(), streamSource);
        }
        String url = null;
        if (model.getStreamUrlIndex() >= 0 && model.getStreamUrlIndex() < streamSources.size()) {
            // TODO don't use the index, but the bandwidth. if the bandwidth does not match, take the closest one
            LOG.debug("{} selected {}", model.getName(), streamSources.get(model.getStreamUrlIndex()));
            url = streamSources.get(model.getStreamUrlIndex()).getMediaPlaylistUrl();
        } else {
            // filter out stream resolutions, which are out of range of the configured min and max
            int minRes = Config.getInstance().getSettings().minimumResolution;
            int maxRes = Config.getInstance().getSettings().maximumResolution;
            List<StreamSource> filteredStreamSources = streamSources.stream()
                    .filter(src -> src.height == 0 || src.height == UNKNOWN || minRes <= src.height)
                    .filter(src -> src.height == 0 || src.height == UNKNOWN || maxRes >= src.height)
                    .collect(Collectors.toList());

            if (filteredStreamSources.isEmpty()) {
                throw new ExecutionException(new RuntimeException("No stream left in playlist"));
            } else {
                LOG.debug("{} selected {}", model.getName(), filteredStreamSources.get(filteredStreamSources.size() - 1));
                url = filteredStreamSources.get(filteredStreamSources.size() - 1).getMediaPlaylistUrl();
            }
        }
        LOG.debug("Segment playlist url {}", url);
        return url;
    }

    protected SegmentPlaylist getNextSegments(String segmentPlaylistUrl) throws IOException, ParseException, PlaylistException {
        Instant start = Instant.now();
        recordingEvents.add(RecordingEvent.of("Playlist request"));
        URL segmentsUrl = new URL(segmentPlaylistUrl);
        Builder builder = new Request.Builder()
                .url(segmentsUrl);
        addHeaders(builder, Optional.ofNullable(model).map(Model::getHttpHeaderFactory).map(HttpHeaderFactory::createSegmentPlaylistHeaders).orElse(new HashMap<>()), model);
        Request request = builder.build();

        try (Response response = client.execute(request, config.getSettings().playlistRequestTimeout)) {
            if (response.isSuccessful()) {
                consecutivePlaylistTimeouts = 0;
                String body = response.body().string();
                if (!body.contains("#EXTINF")) {
                    // no segments, empty playlist
                    return new SegmentPlaylist(segmentPlaylistUrl);
                }

                byte[] bytes = body.getBytes(UTF_8);
                BandwidthMeter.add(bytes.length);
                InputStream inputStream = new ByteArrayInputStream(bytes);
                SegmentPlaylist playList = parsePlaylist(segmentPlaylistUrl, inputStream);
                consecutivePlaylistErrors = 0;
                recordingEvents.add(RecordingEvent.of("Sequence: " + StringUtil.grep(body, "MEDIA-SEQUENCE")));
                recordingEvents.add(RecordingEvent.of("Playlist downloaded in " + (Duration.between(start, Instant.now()).toMillis()) + "ms: "
                        + StringUtil.grep(body, "X-PROGRAM-DATE-TIME")));
                return playList;
            } else {
                recordingEvents.add(RecordingEvent.of("HTTP code " + response.code()));
                throw new HttpException(response.code(), response.message());
            }
        } catch (SocketTimeoutException e) {
            LOG.debug("Playlist request timed out ({}ms) for model {} {} time{}", config.getSettings().playlistRequestTimeout, model,
                    ++consecutivePlaylistTimeouts, (consecutivePlaylistTimeouts > 1) ? 's' : "");
            // times out, return an empty playlist, so that the process can continue without wasting much more time
            recordingEvents.add(RecordingEvent.of("Playlist request timed out " + consecutivePlaylistTimeouts));
            throw new PlaylistTimeoutException(e);
        } catch (Exception e) {
            consecutivePlaylistErrors++;
            throw e;
        }
    }

    private SegmentPlaylist parsePlaylist(String segmentPlaylistUrl, InputStream inputStream) throws IOException, ParseException, PlaylistException {
        PlaylistParser parser = new PlaylistParser(inputStream, Format.EXT_M3U, Encoding.UTF_8, ParsingMode.LENIENT);
        Playlist playlist = parser.parse();
        if (playlist.hasMediaPlaylist()) {
            MediaPlaylist mediaPlaylist = playlist.getMediaPlaylist();
            SegmentPlaylist lsp = new SegmentPlaylist(segmentPlaylistUrl);
            lsp.seq = mediaPlaylist.getMediaSequenceNumber();
            lsp.targetDuration = mediaPlaylist.getTargetDuration();

            List<TrackData> tracks = mediaPlaylist.getTracks();
            for (TrackData trackData : tracks) {
                String uri = trackData.getUri();
                if (!uri.startsWith("http")) {
                    URL context = new URL(segmentPlaylistUrl);
                    uri = new URL(context, uri).toExternalForm();
                }
                lsp.totalDuration += trackData.getTrackInfo().duration;
                lsp.segments.add(uri);
                if (trackData.hasEncryptionData()) {
                    lsp.encrypted = true;
                    EncryptionData data = trackData.getEncryptionData();
                    lsp.encryptionKeyUrl = data.getUri();
                    lsp.encryptionMethod = data.getMethod().getValue();
                }
            }
            lsp.avgSegDuration = lsp.totalDuration / tracks.size();
            return lsp;
        }
        throw new InvalidPlaylistException("Playlist has no media playlist");
    }

    protected void emptyPlaylistCheck(SegmentPlaylist playlist) {
        if(playlist.segments.isEmpty()) {
            playlistEmptyCount++;
            try {
                Thread.sleep(6000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        } else {
            playlistEmptyCount = 0;
        }
        if(playlistEmptyCount == 10) {
            LOG.info("Last 10 playlists were empty for {}. Stopping recording!", getModel());
            internalStop();
        }
    }

    private void handleMissedSegments(SegmentPlaylist playlist, int nextSegmentNumber) throws IOException {
        if (nextSegmentNumber > 0 && playlist.seq > nextSegmentNumber) {
            recordingEvents.add(RecordingEvent.of("Missed segments: "+nextSegmentNumber+" < " + playlist.seq));
            if (DEBUG_HLS) {
                File hlsEventsFile = File.createTempFile("/tmp/rec_evt_" + Instant.now() + "_" + model.getSanitizedNamed(), ".log");
                try (OutputStream outputStream = Files.newOutputStream(hlsEventsFile.toPath(), CREATE, WRITE, TRUNCATE_EXISTING)) {
                    BufferedWriter br = new BufferedWriter(new OutputStreamWriter(outputStream));
                    DateTimeFormatter dtf = DateTimeFormatter.ISO_LOCAL_TIME;
                    for (RecordingEvent recordingEvent : recordingEvents) {
                        ZonedDateTime dateTime = recordingEvent.timestamp.atZone(ZoneId.systemDefault());
                        br.append(dtf.format(dateTime) + " " + model + " " + recordingEvent.message);
                        br.newLine();
                    }
                    br.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void splitRecordingIfNecessary() {
        if (splittingStrategy.splitNecessary(this)) {
            internalStop();
        }
    }

    protected void enqueueNewSegments(SegmentPlaylist playlist, int nextSegmentNumber) throws IOException {
        int skip = nextSegmentNumber - playlist.seq;
        for (String segment : playlist.segments) {
            if (skip > 0) {
                skip--;
            } else {
                URL segmentUrl = new URL(segment);
                String prefix = nf.format(segmentCounter++);
                File tmp = new File(segmentUrl.getFile());
                OutputStream targetStream = getSegmentOutputStream(prefix, tmp.getName());
                SegmentDownload segmentDownload = new SegmentDownload(model, playlist, segmentUrl, client, targetStream);
                execute(segmentDownload);
            }
        }
    }

    private void calculateRescheduleTime() {
        rescheduleTime = beforeLastPlaylistRequest.plusMillis(1000);
        recordingEvents.add(RecordingEvent.of("next playlist download scheduled for " + rescheduleTime.toString()));
    }

    /**
     * Causes the current thread to sleep for a short amount of time.
     * This is used to slow down retries, if something is wrong with the playlist.
     * E.g. HTTP 403 or 404
     */
    protected void waitSomeTime(long waitForMillis) {
        try {
            Thread.sleep(waitForMillis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            if(running) {
                LOG.error("Couldn't sleep. This might mess up the download!");
            }
        }
    }

    protected static void addHeaders(Builder builder, Map<String, String> headers, Model model) {
        headers.putIfAbsent(ACCEPT, "*/*");
        headers.putIfAbsent(ACCEPT_LANGUAGE, Locale.ENGLISH.getLanguage());
        headers.putIfAbsent(USER_AGENT, Config.getInstance().getSettings().httpUserAgent);
        headers.putIfAbsent(CONNECTION, KEEP_ALIVE);
        headers.computeIfAbsent(HttpConstants.ORIGIN, k -> Optional.ofNullable(model).map(Model::getSite).map(Site::getBaseUrl).orElse(null));
        headers.computeIfAbsent(REFERER, k -> Optional.ofNullable(model).map(Model::getSite).map(Site::getBaseUrl).orElse(null));

        for (Entry<String, String> header : headers.entrySet()) {
            builder.header(header.getKey(), header.getValue());
        }
    }

    @Override
    public Model getModel() {
        return model;
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    private static class RecordingEvent {
        Instant timestamp;
        String message;

        public static RecordingEvent of(String message) {
            RecordingEvent evt = new RecordingEvent();
            evt.timestamp = Instant.now();
            evt.message = message;
            return evt;
        }
    }
}
