/*
 * Copyright (C) 2016, 2017 Metrological Group B.V.
 * Copyright (C) 2016, 2017 Igalia S.L
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public License
 * aint with this library; see the file COPYING.LIB.  If not, write to
 * the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include "config.h"
#include "AppendPipeline.h"
#include "AbortableTaskQueue.h"
#include "MediaSourcePrivateGStreamer.h"

#if ENABLE(VIDEO) && USE(GSTREAMER) && ENABLE(MEDIA_SOURCE)

#include "AudioTrackPrivateGStreamer.h"
#include "GStreamerCommon.h"
#include "GStreamerEMEUtilities.h"
#include "GStreamerMediaDescription.h"
#include "GStreamerRegistryScannerMSE.h"
#include "InbandTextTrackPrivateGStreamer.h"
#include "MediaDescription.h"
#include "MediaSampleGStreamer.h"
#include "SourceBufferPrivateGStreamer.h"
#include "VideoTrackPrivateGStreamer.h"
#include <functional>
#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <gst/gst.h>
#include <gst/pbutils/pbutils.h>
#include <gst/video/video.h>
#include <wtf/Condition.h>
#include <wtf/TZoneMallocInlines.h>
#include <wtf/glib/RunLoopSourcePriority.h>
#include <wtf/text/MakeString.h>

GST_DEBUG_CATEGORY_STATIC(webkit_mse_append_pipeline_debug);
#define GST_CAT_DEFAULT webkit_mse_append_pipeline_debug

namespace WebCore {

WTF_MAKE_TZONE_ALLOCATED_IMPL(AppendPipeline);
WTF_MAKE_TZONE_ALLOCATED_IMPL(AppendPipeline::Track);

GType AppendPipeline::s_endOfAppendMetaType = 0;
const GstMetaInfo* AppendPipeline::s_webKitEndOfAppendMetaInfo = nullptr;
std::once_flag AppendPipeline::s_staticInitializationFlag;

struct EndOfAppendMeta {
    GstMeta base;
    static gboolean init(GstMeta*, void*, GstBuffer*) { return TRUE; }
    static gboolean transform(GstBuffer*, GstMeta*, GstBuffer*, GQuark, void*) { g_return_val_if_reached(FALSE); }
    static void free(GstMeta*, GstBuffer*) { }
};

void AppendPipeline::staticInitialization()
{
    ASSERT(isMainThread());
    GST_DEBUG_CATEGORY_INIT(webkit_mse_append_pipeline_debug, "webkitmseappendpipeline", 0, "WebKit MSE AppendPipeline");

    const char* tags[] = { nullptr };
    s_endOfAppendMetaType = gst_meta_api_type_register("WebKitEndOfAppendMetaAPI", tags);
    s_webKitEndOfAppendMetaInfo = gst_meta_register(s_endOfAppendMetaType, "WebKitEndOfAppendMeta", sizeof(EndOfAppendMeta), EndOfAppendMeta::init, EndOfAppendMeta::free, EndOfAppendMeta::transform);
}

#if !LOG_DISABLED
static GstPadProbeReturn appendPipelinePadProbeDebugInformation(GstPad*, GstPadProbeInfo*, struct PadProbeInformation*);
#endif

#if ENABLE(ENCRYPTED_MEDIA)
static GstPadProbeReturn appendPipelineAppsinkPadEventProbe(GstPad*, GstPadProbeInfo*, struct PadProbeInformation*);
#endif
static GstPadProbeReturn appendPipelineDemuxerBlackHolePadProbe(GstPad*, GstPadProbeInfo*, gpointer);
static GstPadProbeReturn matroskademuxForceSegmentStartToEqualZero(GstPad*, GstPadProbeInfo*, void*);
static GRefPtr<GstElement> createOptionalParserForFormat(GstBin*, const String&, const GstCaps*);
static GRefPtr<GstElement> createOptionalEncoderForFormat(GstBin*, const String&, const GstCaps*);

// Wrapper for gst_element_set_state() that emits a critical if the state change fails or is not synchronous.
static void assertedElementSetState(GstElement* element, GstState desiredState)
{
    GstState oldState;
    gst_element_get_state(element, &oldState, nullptr, 0);

    GstStateChangeReturn result = gst_element_set_state(element, desiredState);

    GstState newState;
    gst_element_get_state(element, &newState, nullptr, 0);

    if (desiredState != newState || result != GST_STATE_CHANGE_SUCCESS) {
        GST_ERROR_OBJECT(element, "AppendPipeline state change failed (returned %s): %s -> %s (expected %s)", gst_element_state_change_return_get_name(result), gst_element_state_get_name(oldState), gst_element_state_get_name(newState), gst_element_state_get_name(desiredState));
        ASSERT_NOT_REACHED();
    }
}

AppendPipeline::AppendPipeline(SourceBufferPrivateGStreamer& sourceBufferPrivate, MediaPlayerPrivateGStreamerMSE& playerPrivate)
    : m_sourceBufferPrivate(sourceBufferPrivate)
    , m_playerPrivate(&playerPrivate)
    , m_wasBusAlreadyNotifiedOfAvailableSamples(false)
{
    ASSERT(isMainThread());
    std::call_once(s_staticInitializationFlag, AppendPipeline::staticInitialization);

    GST_DEBUG_OBJECT(m_playerPrivate->pipeline(), "Creating AppendPipeline (%p)", this);

    // FIXME: give a name to the pipeline, maybe related with the track it's managing.
    // The track name is still unknown at this time, though.
    static size_t appendPipelineCount = 0;
    String pipelineName = makeString("append-pipeline-"_s,
        makeStringByReplacingAll(m_sourceBufferPrivate.type().containerType(), '/', '-'), '-', appendPipelineCount++);
    m_pipeline = gst_pipeline_new(pipelineName.utf8().data());
    registerActivePipeline(m_pipeline);
    connectSimpleBusMessageCallback(m_pipeline.get());

    auto bus = adoptGRef(gst_pipeline_get_bus(GST_PIPELINE(m_pipeline.get())));
    gst_bus_enable_sync_message_emission(bus.get());

    g_signal_connect(bus.get(), "sync-message::error", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) {
        appendPipeline->handleErrorSyncMessage(message);
    }), this);
    g_signal_connect(bus.get(), "sync-message::need-context", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) {
        appendPipeline->handleNeedContextSyncMessage(message);
    }), this);

    // We assign the created instances here instead of adoptRef() because gst_bin_add_many()
    // below will already take the initial reference and we need an additional one for us.
    m_appsrc = makeGStreamerElement("appsrc", nullptr);

    GRefPtr<GstPad> appsrcPad = adoptGRef(gst_element_get_static_pad(m_appsrc.get(), "src"));
    gst_pad_add_probe(appsrcPad.get(), GST_PAD_PROBE_TYPE_BUFFER, [](GstPad*, GstPadProbeInfo* padProbeInfo, void* userData) {
        return static_cast<AppendPipeline*>(userData)->appsrcEndOfAppendCheckerProbe(padProbeInfo);
    }, this, nullptr);

    const String& type = m_sourceBufferPrivate.type().containerType();
    GST_DEBUG_OBJECT(pipeline(), "SourceBuffer containerType: %s", type.utf8().data());

    if (type.endsWith("mp4"_s) || type.endsWith("aac"_s)) {
        m_demux = makeGStreamerElement("qtdemux", nullptr);
        m_typefind = makeGStreamerElement("identity", nullptr);
        GRefPtr<GstCaps> caps = adoptGRef(gst_caps_new_simple("video/quicktime", "variant", G_TYPE_STRING, "mse-bytestream", NULL));
        gst_app_src_set_caps(GST_APP_SRC(m_appsrc.get()), caps.get());
    } else if (type.endsWith("webm"_s)) {
        m_demux = makeGStreamerElement("matroskademux", nullptr);
        m_typefind = makeGStreamerElement("identity", nullptr);
    } else if (type == "audio/mpeg"_s) {
        m_demux = makeGStreamerElement("identity", nullptr);
        m_typefind = makeGStreamerElement("typefind", nullptr);
    } else
        ASSERT_NOT_REACHED();

#if !LOG_DISABLED
    GRefPtr<GstPad> demuxerPad = adoptGRef(gst_element_get_static_pad(m_demux.get(), "sink"));
    m_demuxerDataEnteringPadProbeInformation.appendPipeline = this;
    m_demuxerDataEnteringPadProbeInformation.description = "demuxer data entering";
    m_demuxerDataEnteringPadProbeInformation.probeId = gst_pad_add_probe(demuxerPad.get(), GST_PAD_PROBE_TYPE_BUFFER, reinterpret_cast<GstPadProbeCallback>(appendPipelinePadProbeDebugInformation), &m_demuxerDataEnteringPadProbeInformation, nullptr);
#endif

    String elementClass = unsafeSpan(gst_element_get_metadata(m_demux.get(), GST_ELEMENT_METADATA_KLASS));
    auto classifiers = elementClass.split('/');
    if (classifiers.contains("Demuxer"_s)) {
        // These signals won't outlive the lifetime of `this`.
        g_signal_connect_swapped(m_demux.get(), "no-more-pads", G_CALLBACK(+[](AppendPipeline* appendPipeline) {
            ASSERT(!isMainThread());
            GST_DEBUG_OBJECT(appendPipeline->pipeline(), "Posting no-more-pads task to main thread");
            appendPipeline->m_taskQueue.enqueueTaskAndWait<AbortableTaskQueue::Void>([appendPipeline]() {
                appendPipeline->didReceiveInitializationSegment();
                return AbortableTaskQueue::Void();
            });
        }), this);
    } else {
        GRefPtr<GstPad> identitySrcPad = adoptGRef(gst_element_get_static_pad(m_demux.get(), "src"));
        gst_pad_add_probe(identitySrcPad.get(), GST_PAD_PROBE_TYPE_BUFFER, reinterpret_cast<GstPadProbeCallback>(
            +[](GstPad *pad, GstPadProbeInfo*, AppendPipeline* appendPipeline) {
                GRefPtr<GstCaps> caps = adoptGRef(gst_pad_get_current_caps(pad));
                if (!caps)
                    return GST_PAD_PROBE_DROP;
                appendPipeline->m_taskQueue.enqueueTaskAndWait<AbortableTaskQueue::Void>([appendPipeline]() {
                    appendPipeline->didReceiveInitializationSegment();
                    return AbortableTaskQueue::Void();
                });
                return GST_PAD_PROBE_REMOVE;
            }
        ), this, nullptr);
    }

    // Add_many will take ownership of a reference. That's why we used an assignment before.
    gst_bin_add_many(GST_BIN(m_pipeline.get()), m_appsrc.get(), m_typefind.get(), m_demux.get(), nullptr);
    gst_element_link_many(m_appsrc.get(), m_typefind.get(), m_demux.get(), nullptr);

    assertedElementSetState(m_pipeline.get(), GST_STATE_PLAYING);
}

AppendPipeline::~AppendPipeline()
{
    GST_DEBUG_OBJECT(pipeline(), "Destructing AppendPipeline (%p)", this);
    ASSERT(isMainThread());

    // Forget all pending tasks and unblock the streaming thread if it was blocked.
    m_taskQueue.startAborting();

    // Disconnect all synchronous event handlers and probes susceptible of firing from the main thread
    // when changing the pipeline state.

    if (m_pipeline) {
        auto bus = adoptGRef(gst_pipeline_get_bus(GST_PIPELINE(m_pipeline.get())));
        ASSERT(bus);
        g_signal_handlers_disconnect_by_data(bus.get(), this);
        gst_bus_disable_sync_message_emission(bus.get());
        disconnectSimpleBusMessageCallback(m_pipeline.get());
    }

    if (m_demux) {
#if !LOG_DISABLED
        GRefPtr<GstPad> demuxerPad = adoptGRef(gst_element_get_static_pad(m_demux.get(), "sink"));
        gst_pad_remove_probe(demuxerPad.get(), m_demuxerDataEnteringPadProbeInformation.probeId);
#endif

        g_signal_handlers_disconnect_by_data(m_demux.get(), this);
    }

    for (std::unique_ptr<Track>& track : m_tracks) {
        GRefPtr<GstPad> appsinkPad = adoptGRef(gst_element_get_static_pad(track->appsink.get(), "sink"));
        g_signal_handlers_disconnect_by_data(appsinkPad.get(), this);
        g_signal_handlers_disconnect_by_data(track->appsink.get(), this);
#if !LOG_DISABLED
        gst_pad_remove_probe(appsinkPad.get(), track->appsinkDataEnteringPadProbeInformation.probeId);
#endif
#if ENABLE(ENCRYPTED_MEDIA)
        gst_pad_remove_probe(appsinkPad.get(), track->appsinkPadEventProbeInformation.probeId);
#endif
        m_sourceBufferPrivate.tryUnregisterTrackId(track->trackId);
    }

    // We can tear down the pipeline safely now.
    if (m_pipeline) {
        unregisterPipeline(m_pipeline);
        gst_element_set_state(m_pipeline.get(), GST_STATE_NULL);
    }
}

void AppendPipeline::handleErrorConditionFromStreamingThread()
{
    ASSERT(!isMainThread());
    m_taskQueue.enqueueTaskAndWait<AbortableTaskQueue::Void>([this]() {
        m_errorReceived = true;
        // appendParsingFailed() will cause resetParserState() to be called.
        m_sourceBufferPrivate.appendParsingFailed();
        return AbortableTaskQueue::Void();
    });
}

void AppendPipeline::handleErrorSyncMessage([[maybe_unused]] GstMessage* message)
{
    ASSERT(!isMainThread());
    GST_WARNING_OBJECT(pipeline(), "Demuxing error: %" GST_PTR_FORMAT, message);
    handleErrorConditionFromStreamingThread();
    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, "demuxing-error");
}

GstPadProbeReturn AppendPipeline::appsrcEndOfAppendCheckerProbe(GstPadProbeInfo* padProbeInfo)
{
    ASSERT(!isMainThread());
    m_streamingThread = &Thread::current();

    GstBuffer* buffer = GST_BUFFER(padProbeInfo->data);
    ASSERT(GST_IS_BUFFER(buffer));

    GST_TRACE_OBJECT(pipeline(), "Buffer entered appsrcEndOfAppendCheckerProbe: %" GST_PTR_FORMAT, buffer);

    EndOfAppendMeta* endOfAppendMeta = reinterpret_cast<EndOfAppendMeta*>(gst_buffer_get_meta(buffer, s_endOfAppendMetaType));
    if (!endOfAppendMeta) {
        // Normal buffer, nothing to do.
        return GST_PAD_PROBE_OK;
    }

    GST_TRACE_OBJECT(pipeline(), "Posting end-of-append task to the main thread");
    m_taskQueue.enqueueTask([this]() {
        handleEndOfAppend();
    });
    return GST_PAD_PROBE_DROP;
}

void AppendPipeline::handleNeedContextSyncMessage(GstMessage* message)
{
    // MediaPlayerPrivateGStreamerBase will take care of setting up encryption.
    m_playerPrivate->handleNeedContextMessage(message);
}

std::tuple<GRefPtr<GstCaps>, AppendPipeline::StreamType, FloatSize> AppendPipeline::parseDemuxerSrcPadCaps(GstCaps* demuxerSrcPadCaps)
{
    ASSERT(isMainThread());

    GRefPtr<GstCaps> parsedCaps = demuxerSrcPadCaps;
    StreamType streamType = StreamType::Unknown;
    FloatSize presentationSize;

    auto originalMediaType = capsMediaType(demuxerSrcPadCaps);
    auto& gstRegistryScanner = GStreamerRegistryScannerMSE::singleton();
    if (!gstRegistryScanner.isCodecSupported(GStreamerRegistryScanner::Configuration::Decoding, originalMediaType.toStringWithoutCopying())) {
        streamType = StreamType::Invalid;
    } else if (doCapsHaveType(demuxerSrcPadCaps, GST_VIDEO_CAPS_TYPE_PREFIX)) {
        presentationSize = getVideoResolutionFromCaps(demuxerSrcPadCaps).value_or(FloatSize());
        streamType = StreamType::Video;
    } else {
        if (doCapsHaveType(demuxerSrcPadCaps, GST_AUDIO_CAPS_TYPE_PREFIX))
            streamType = StreamType::Audio;
        else if (doCapsHaveType(demuxerSrcPadCaps, GST_TEXT_CAPS_TYPE_PREFIX))
            streamType = StreamType::Text;
    }

    return { WTFMove(parsedCaps), streamType, WTFMove(presentationSize) };
}

void AppendPipeline::appsinkCapsChanged(Track& track)
{
    ASSERT(isMainThread());

    // Consume any pending samples with the previous caps.
    consumeAppsinksAvailableSamples();

    GRefPtr<GstPad> pad = adoptGRef(gst_element_get_static_pad(track.appsink.get(), "sink"));
    GRefPtr<GstCaps> caps = adoptGRef(gst_pad_get_current_caps(pad.get()));

    if (!caps)
        return;

    // If this is not the first time we're parsing an initialization segment, fail if the track
    // has a different codec or type (e.g. if we were previously demuxing an audio stream and
    // someone appends a video stream).
    auto currentMediaTypeView = capsMediaType(caps.get());
    auto trackMediaTypeView = capsMediaType(track.caps.get());
    if (track.finalCaps && currentMediaTypeView != trackMediaTypeView) {
#ifndef GST_DISABLE_GST_DEBUG
        auto currentMediaType = currentMediaTypeView.utf8();
        auto trackMediaType = trackMediaTypeView.utf8();
        GST_WARNING_OBJECT(pipeline(), "Track received incompatible caps, received '%s' for a track previously handling '%s'. Erroring out.", currentMediaType.data(), trackMediaType.data());
#endif
        m_sourceBufferPrivate.appendParsingFailed();
        return;
    }

    if (doCapsHaveType(caps.get(), GST_VIDEO_CAPS_TYPE_PREFIX)) {
        if (auto size = getVideoResolutionFromCaps(caps.get()))
            track.presentationSize = *size;
    }

    if (track.caps != caps)
        track.caps = WTFMove(caps);
}

void AppendPipeline::handleEndOfAppend()
{
    ASSERT(isMainThread());
    consumeAppsinksAvailableSamples();
    GST_TRACE_OBJECT(pipeline(), "Notifying SourceBufferPrivate the append is complete");
    sourceBufferPrivate().didReceiveAllPendingSamples();
}

void AppendPipeline::appsinkNewSample(const Track& track, GRefPtr<GstSample>&& sample)
{
    ASSERT(isMainThread());

    if (UNLIKELY(!gst_sample_get_buffer(sample.get()))) {
        GST_WARNING_OBJECT(pipeline(), "Received sample without buffer from appsink.");
        return;
    }

    auto* buffer = gst_sample_get_buffer(sample.get());
    if (!GST_BUFFER_PTS_IS_VALID(buffer)) {
        // When demuxing Vorbis, matroskademux creates several PTS-less frames with header information. We don't need those.
        GST_DEBUG_OBJECT(pipeline(), "Ignoring sample without PTS: %" GST_PTR_FORMAT, buffer);
        return;
    }

    auto mediaSample = MediaSampleGStreamer::create(WTFMove(sample), track.presentationSize, track.trackId);

    GST_TRACE_OBJECT(pipeline(), "append: trackId=%" PRIu64 " PTS=%s DTS=%s DUR=%s presentationSize=%.0fx%.0f",
        mediaSample->trackID(),
        mediaSample->presentationTime().toString().utf8().data(),
        mediaSample->decodeTime().toString().utf8().data(),
        mediaSample->duration().toString().utf8().data(),
        mediaSample->presentationSize().width(), mediaSample->presentationSize().height());

    // Hack, rework when GStreamer >= 1.16 becomes a requirement:
    // We're not applying edit lists. GStreamer < 1.16 doesn't emit the correct segments to do so.
    // GStreamer fix in https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/-/commit/c2a0da8096009f0f99943f78dc18066965be60f9
    // Also, in order to apply them we would need to convert the timestamps to stream time, which we're not currently
    // doing for consistency between GStreamer versions.
    //
    // In consequence, the timestamps we're handling here are unedited track time. In track time, the first sample is
    // guaranteed to have DTS == 0, but in the case of streams with B-frames, often PTS > 0. Edit lists fix this by
    // offsetting all timestamps by that amount in movie time, but we can't do that if we don't have access to them.
    // (We could assume the track PTS of the sample with track DTS = 0 is the offset, but we don't have any guarantee
    // we will get appended that sample first, or ever).
    //
    // Because a track presentation time starting at some close to zero, but not exactly zero time can cause unexpected
    // results for applications, we extend the duration of this first sample to the left so that it starts at zero.
    if (mediaSample->decodeTime() == MediaTime::zeroTime() && mediaSample->presentationTime() > MediaTime::zeroTime()
        && mediaSample->presentationTime() <= MediaTime(1, 10)
        && mediaSample->isSync()) {
        GST_DEBUG_OBJECT(pipeline(), "Extending first sample to make it start at PTS=0");
        mediaSample->extendToTheBeginning();
    }

    m_sourceBufferPrivate.didReceiveSample(mediaSample.get());
}

void AppendPipeline::didReceiveInitializationSegment()
{
    ASSERT(isMainThread());

    bool isFirstInitializationSegment = !m_hasReceivedFirstInitializationSegment;

    SourceBufferPrivateClient::InitializationSegment initializationSegment;

    gint64 timeLength = 0;
    if (gst_element_query_duration(m_demux.get(), GST_FORMAT_TIME, &timeLength)
        && static_cast<guint64>(timeLength) != GST_CLOCK_TIME_NONE)
        initializationSegment.duration = MediaTime(GST_TIME_AS_USECONDS(timeLength), G_USEC_PER_SEC);
    else
        initializationSegment.duration = MediaTime::positiveInfiniteTime();

    if (isFirstInitializationSegment) {
        // Create a Track object per pad.
        for (GstPad* pad : GstIteratorAdaptor<GstPad>(GUniquePtr<GstIterator>(gst_element_iterate_src_pads(m_demux.get())))) {
            auto [createTrackResult, track] = tryCreateTrackFromPad(pad);
            if (createTrackResult == CreateTrackResult::AppendParsingFailed) {
                // appendParsingFailed() will immediately cause a resetParserState() which will stop demuxing, then the
                // AppendPipeline will be destroyed.
                m_sourceBufferPrivate.appendParsingFailed();
                return;
            }
            if (track)
                linkPadWithTrack(pad, *track);
        }
    } else {
        UncheckedKeyHashSet<String> videoPadStreamIDs;
        UncheckedKeyHashSet<String> audioPadStreamIDs;
        UncheckedKeyHashSet<String> textPadStreamIDs;
        for (auto pad : GstIteratorAdaptor<GstPad>(GUniquePtr<GstIterator>(gst_element_iterate_src_pads(m_demux.get())))) {
            auto [parsedCaps, streamType, presentationSize] = parseDemuxerSrcPadCaps(adoptGRef(gst_pad_get_current_caps(pad)).get());
            UNUSED_VARIABLE(parsedCaps);
            UNUSED_VARIABLE(presentationSize);
            String streamID = String::fromLatin1(GUniquePtr<char>(gst_pad_get_stream_id(pad)).get());
            if (streamType == StreamType::Audio)
                audioPadStreamIDs.add(WTFMove(streamID));
            else if (streamType == StreamType::Video)
                videoPadStreamIDs.add(WTFMove(streamID));
            else if (streamType == StreamType::Text)
                textPadStreamIDs.add(WTFMove(streamID));
        }

        unsigned videoTracksCount = 0;
        unsigned audioTracksCount = 0;
        unsigned textTracksCount = 0;
        bool doAudioTrackStreamIDsMatch = true;
        bool doVideoTrackStreamIDsMatch = true;
        bool doTextTrackStreamIDsMatch = true;
        for (const auto& track : m_tracks) {
            GUniquePtr<char> streamIDAsCharacters(gst_pad_get_stream_id(track->entryPad.get()));
            String streamID = String::fromLatin1(streamIDAsCharacters.get());
            if (track->streamType == StreamType::Audio) {
                audioTracksCount++;
                if (streamID.isEmpty() || !audioPadStreamIDs.contains(streamID))
                    doAudioTrackStreamIDsMatch = false;
            } else if (track->streamType == StreamType::Video) {
                videoTracksCount++;
                if (streamID.isEmpty() || !videoPadStreamIDs.contains(streamID))
                    doVideoTrackStreamIDsMatch = false;
            } else if (track->streamType == StreamType::Text) {
                textTracksCount++;
                if (streamID.isEmpty() || !textPadStreamIDs.contains(streamID))
                    doTextTrackStreamIDsMatch = false;
            }
        }

        // Step 3 of the MSE spec append initialization segment algorithm.
        // 4.5.7.3.1: Verify the following properties. If any of the checks fail then run the append error algorithm and abort these steps.
        // 4.5.7.3.1.1: The number of audio, video, and text tracks match what was in the first initialization segment.
        bool doTracksMatch = audioPadStreamIDs.size() == audioTracksCount && videoPadStreamIDs.size() == videoTracksCount && textPadStreamIDs.size() == textTracksCount;
        // 4.5.7.3.1.2: If more than one track for a single type are present (e.g., 2 audio tracks), then the Track IDs match the ones in the first initialization segment.
        if (doTracksMatch && audioTracksCount > 1)
            doTracksMatch = doAudioTrackStreamIDsMatch;
        if (doTracksMatch && videoTracksCount > 1)
            doTracksMatch = doVideoTrackStreamIDsMatch;
        if (doTracksMatch && textTracksCount > 1)
            doTracksMatch = doTextTrackStreamIDsMatch;
        if (!doTracksMatch) {
            GST_WARNING_OBJECT(pipeline(), "New demuxed stream topology doesn't match the existing tracks topology");
            m_sourceBufferPrivate.appendParsingFailed();
            return;
        }

        // Link pads to existing Track objects that don't have a linked pad yet. Existing linked
        // tracks are recycled if their stream type matches the new demuxer source pads.
        for (GstPad* pad : GstIteratorAdaptor<GstPad>(GUniquePtr<GstIterator>(gst_element_iterate_src_pads(m_demux.get())))) {
            if (!recycleTrackForPad(pad)) {
                GST_WARNING_OBJECT(pipeline(), "Can't match pad to existing tracks in the AppendPipeline: %" GST_PTR_FORMAT, pad);
                m_sourceBufferPrivate.appendParsingFailed();
                return;
            }
        }
    }

    for (std::unique_ptr<Track>& track : m_tracks) {
#ifndef GST_DISABLE_GST_DEBUG
        GST_DEBUG_OBJECT(pipeline(), "Adding track to initialization with segment type %s, id %" PRIu64 ".", streamTypeToString(track->streamType), track->trackId);
#endif // GST_DISABLE_GST_DEBUG
        switch (track->streamType) {
        case Audio: {
            ASSERT(track->webKitTrack);
            SourceBufferPrivateClient::InitializationSegment::AudioTrackInformation info;
            info.track = static_cast<AudioTrackPrivateGStreamer*>(track->webKitTrack.get());
            info.description = GStreamerMediaDescription::create(track->finalCaps);
            initializationSegment.audioTracks.append(info);
            break;
        }
        case Video: {
            ASSERT(track->webKitTrack);
            SourceBufferPrivateClient::InitializationSegment::VideoTrackInformation info;
            info.track = static_cast<VideoTrackPrivateGStreamer*>(track->webKitTrack.get());
            info.description = GStreamerMediaDescription::create(track->finalCaps);
            initializationSegment.videoTracks.append(info);
            break;
        }
        default:
            GST_ERROR("Unsupported stream type or codec");
            break;
        }
    }

    if (isFirstInitializationSegment) {
        for (std::unique_ptr<Track>& track : m_tracks) {
            if (track->streamType == StreamType::Video) {
                GST_DEBUG_OBJECT(pipeline(), "Setting initial video size to that of track with id '%" PRIu64 "', %gx%g.",
                    track->trackId, static_cast<double>(track->presentationSize.width()), static_cast<double>(track->presentationSize.height()));
                m_playerPrivate->setInitialVideoSize(track->presentationSize);
                break;
            }
        }
    }

    m_hasReceivedFirstInitializationSegment = true;
    GST_DEBUG_OBJECT(pipeline(), "Notifying SourceBuffer of initialization segment.");
    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, "append-pipeline-received-init-segment");
    m_sourceBufferPrivate.didReceiveInitializationSegment(WTFMove(initializationSegment));
}

void AppendPipeline::consumeAppsinksAvailableSamples()
{
    ASSERT(isMainThread());

    GRefPtr<GstSample> sample;
    [[maybe_unused]] int batchedSampleCount = 0;
    for (std::unique_ptr<Track>& track : m_tracks) {
        while ((sample = adoptGRef(gst_app_sink_try_pull_sample(GST_APP_SINK(track->appsink.get()), 0)))) {
            appsinkNewSample(*track, WTFMove(sample));
            batchedSampleCount++;
        }
    }

    GST_TRACE_OBJECT(pipeline(), "batchedSampleCount = %d", batchedSampleCount);
}

void AppendPipeline::resetParserState()
{
    ASSERT(isMainThread());

    if (!m_pipeline)
        return;

    // This function restores the GStreamer pipeline to the same state it was when the AppendPipeline constructor
    // finished. All previously enqueued data is lost and the demuxer is flushed, but retains the configuration from the
    // last received init segment, in accordance to the spec.

    // Unlock the streaming thread.
    m_taskQueue.startAborting();

    // Flush approach requires these GStreamer patches, shipped in 1.24:
    // https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/4101.
    // https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/4199.
    if (webkitGstCheckVersion(1, 24, 0)) {
        GST_DEBUG_OBJECT(pipeline(), "Handling resetParserState() in AppendPipeline by flushing the pipeline");
        gst_element_send_event(m_appsrc.get(), gst_event_new_flush_start());
        gst_element_send_event(m_appsrc.get(), gst_event_new_flush_stop(true));

        GstSegment segment;
        gst_segment_init(&segment, GST_FORMAT_BYTES);
        gst_element_send_event(m_appsrc.get(), gst_event_new_segment(&segment));
    } else {
        GST_DEBUG_OBJECT(pipeline(), "Handling resetParserState() in AppendPipeline by resetting the pipeline");
        // Reset the state of all elements in the pipeline.
        assertedElementSetState(m_pipeline.get(), GST_STATE_READY);

        // Set the pipeline to PLAYING so that it can be used again.
        assertedElementSetState(m_pipeline.get(), GST_STATE_PLAYING);
    }

    // All processing related to the previous append has been aborted and the pipeline is idle.
    // We can listen again to new requests coming from the streaming thread.
    m_taskQueue.finishAborting();

#if (!(LOG_DISABLED || defined(GST_DISABLE_GST_DEBUG)))
    {
        static unsigned i = 0;
        // This is here for debugging purposes. It does not make sense to have it as class member.
        String dotFileName = makeString("reset-pipeline-"_s, ++i);
        gst_debug_bin_to_dot_file(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
    }
#endif
}

void AppendPipeline::stopParser()
{
    ASSERT(isMainThread());
    GST_DEBUG_OBJECT(pipeline(), "Stopping parser");

    // Forget all pending tasks and unblock the streaming thread if it was blocked.
    m_taskQueue.startAborting();

    // Reset the state of all elements in the pipeline.
    assertedElementSetState(m_pipeline.get(), GST_STATE_READY);

    m_taskQueue.finishAborting();
}

void AppendPipeline::pushNewBuffer(GRefPtr<GstBuffer>&& buffer)
{
    GST_TRACE_OBJECT(pipeline(), "pushing data buffer %" GST_PTR_FORMAT, buffer.get());
    GstFlowReturn pushDataBufferRet = gst_app_src_push_buffer(GST_APP_SRC(m_appsrc.get()), buffer.leakRef());
    // Pushing buffers to appsrc can only fail if the appsrc is flushing, in EOS or stopped. Neither of these should
    // be true at this point.
    if (pushDataBufferRet != GST_FLOW_OK) {
        GST_ERROR_OBJECT(pipeline(), "Failed to push data buffer into appsrc.");
        ASSERT_NOT_REACHED();
    }

    // Push an additional empty buffer that marks the end of the append.
    // This buffer is detected and consumed by appsrcEndOfAppendCheckerProbe(), which uses it to signal the successful
    // completion of the append.
    //
    // This works based on how push mode scheduling works in GStreamer. Note there is a single streaming thread for the
    // AppendPipeline, and within a stream (the portion of a pipeline covered by the same streaming thread, in this case
    // the whole pipeline) a buffer is guaranteed not to be processed by downstream until processing of the previous
    // buffer has completed.

    GstBuffer* endOfAppendBuffer = gst_buffer_new();
    gst_buffer_add_meta(endOfAppendBuffer, s_webKitEndOfAppendMetaInfo, nullptr);

    GST_TRACE_OBJECT(pipeline(), "pushing end-of-append buffer %" GST_PTR_FORMAT, endOfAppendBuffer);
    GstFlowReturn pushEndOfAppendBufferRet = gst_app_src_push_buffer(GST_APP_SRC(m_appsrc.get()), endOfAppendBuffer);
    if (pushEndOfAppendBufferRet != GST_FLOW_OK) {
        GST_ERROR_OBJECT(pipeline(), "Failed to push end-of-append buffer into appsrc.");
        ASSERT_NOT_REACHED();
    }
}

void AppendPipeline::handleAppsinkNewSampleFromStreamingThread(GstElement*)
{
    ASSERT(!isMainThread());
    if (&Thread::current() != m_streamingThread) {
        // m_streamingThreadId has been initialized in appsrcEndOfAppendCheckerProbe().
        // For a buffer to reach the appsink, a buffer must have passed through appsrcEndOfAppendCheckerProbe() first.
        // This error will only raise if someone modifies the pipeline to include more than one streaming thread or
        // removes the appsrcEndOfAppendCheckerProbe(). Either way, the end-of-append detection would be broken.
        // AppendPipeline should have only one streaming thread. Otherwise we can't detect reliably when an appends has
        // been demuxed completely.;
        GST_ERROR_OBJECT(pipeline(), "Appsink received a sample in a different thread than appsrcEndOfAppendCheckerProbe run.");
        ASSERT_NOT_REACHED();
    }

    if (!m_wasBusAlreadyNotifiedOfAvailableSamples.test_and_set()) {
        GST_TRACE_OBJECT(pipeline(), "Posting appsink-new-sample task to the main thread");
        m_taskQueue.enqueueTask([this]() {
            m_wasBusAlreadyNotifiedOfAvailableSamples.clear();
            consumeAppsinksAvailableSamples();
        });
    }
}

GRefPtr<GstElement> createOptionalParserForFormat([[maybe_unused]] GstBin* bin, const String& parserName, const GstCaps* caps)
{
    // Parser elements have either or both of two functions:
    //
    // a) Framing: Several popular formats (notably MPEG Audio) can be used without a container.
    //    MSE supports such formats when operating in "sequence" mode. When using these formats,
    //    the parser is an essential element, as it receives buffers of arbitrary byte sizes
    //    and identifies where each frame starts and ends, splitting them into separate GstBuffer
    //    objects, and reassembling frames that were split between two appends.
    //
    // b) Metadata filling: Even when framing is taken care of by a container, sometimes there is
    //    important metadata missing. This may be the case because the container format does not
    //    require such metadata, or it may be because of broken files. Either way, parsers allow
    //    us to recover potentially missing metadata from the binary contents of audio or video
    //    frames.
    //
    // NOTE: Please add and keep comments updated with the rationale for each parser.

    GstStructure* structure = gst_caps_get_structure(caps, 0);
    auto mediaType = gstStructureGetName(structure);
    // Since parsers are not needed in every case, we can use an identity element as pass-through
    // parser for cases where a parser is not needed, making the management of elements and pads
    // more orthogonal.
    const char* elementClass = "identity";

    if (mediaType == "audio/x-opus"_s) {
        // Necessary for: metadata filling.
        // Frame durations are optional in Matroska/WebM. Although frame durations are not required
        // for regular playback, they're necessary for MSE, especially handling replacement of frames
        // during quality changes.
        // An example of an Opus audio file lacking durations is car_opus_low.webm
        // https://storage.googleapis.com/ytlr-cert.appspot.com/test/materials/media/car_opus_low.webm
        elementClass = "opusparse";
    } else if (mediaType == "video/x-h264"_s) {
        // Necessary for: metadata filling.
        // Some dubiously muxed content lacks the bit specifying what frames are key frames or not.
        // Without this bit, seeks will most often than not cause corrupted output in the decoder,
        // as the browser will be unaware of any dependencies of those frames and they won't be fed
        // to the decoder.
        // An example of such a stream: http://orange-opensource.github.io/hasplayer.js/1.2.0/player.html?url=http://playready.directtaps.net/smoothstreaming/SSWSS720H264/SuperSpeedway_720.ism/Manifest
        elementClass = "h264parse";
    } else if (mediaType == "audio/mpeg"_s) {
        // Necessary for: framing.
        // The Media Source Extensions Byte Stream Format Registry includes MPEG Audio Byte Stream Format
        // as the (as of writing) only one spec-defined format that has the "Generate Timestamps Flag" set
        // to false, i.e. is used without a demuxer, in "sequence" mode.
        // We need a parser to take care of extracting the frames from the byte stream.
        int mpegVersion = gstStructureGet<int>(structure, "mpegversion"_s).value_or(0);
        switch (mpegVersion) {
        case 1:
            // MPEG-1 Part 3 Audio (ISO 11172-3) Layer I -- MP1, archaic
            // MPEG-1 Part 3 Audio (ISO 11172-3) Layer II -- MP2, common in audio broadcasting, e.g. DVB
            // MPEG-1 Part 3 Audio (ISO 11172-3) Layer III -- MP3, the only one of the three most people actually know
            elementClass = "mpegaudioparse";
            break;
        case 2:
            // MPEG-2 Part 7 Advanced Audio Coding (ISO 13818-7) -- MPEG-2 AAC, the original AAC format, widely used,
            // has extensions retrofitted.
        case 4:
            // MPEG-4 Part 3 Audio (ISO 14496-3) -- MPEG-4 Audio, which more often than not also contains AAC audio,
            // defines several extensions to the original AAC, also widely used.
            // Not to be confused with the MP4 file format, which is a container format, not an audio stream format,
            // and can incidentally contain MPEG-4 audio.
            elementClass = "aacparse";
            break;
        default:
            GST_WARNING_OBJECT(bin, "Unsupported audio mpeg caps: %" GST_PTR_FORMAT, caps);
        }
    } else if (mediaType == "video/x-vp9"_s) {
        // Necessary for: metadata filling.
        // Without this parser the codec string set on the corresponding video track will be incomplete.
        elementClass = "vp9parse";
    }

    GST_DEBUG_OBJECT(bin, "Creating %s parser for stream with caps %" GST_PTR_FORMAT, elementClass, caps);
    GRefPtr<GstElement> result(makeGStreamerElement(elementClass, parserName.ascii().data()));
    if (!result && g_strcmp0(elementClass, "identity")) {
        GST_WARNING_OBJECT(bin, "Couldn't create %s, there might be problems processing some MSE streams. "
            "Continue at your own risk and consider adding %s to your build.", elementClass, elementClass);
        result = makeGStreamerElement("identity", parserName.ascii().data());
    }
    return result;
}

GRefPtr<GstElement> createOptionalEncoderForFormat([[maybe_unused]] GstBin* bin, const String& encoderName, const GstCaps* caps)
{
    GstStructure* structure = gst_caps_get_structure(caps, 0);
    auto mediaType = gstStructureGetName(structure);

    const char* elementClass = "identity";

    if (mediaType == "text/x-raw"_s)
        elementClass = "webvttenc";

    GST_DEBUG_OBJECT(bin, "Creating %s encoder for stream with caps %" GST_PTR_FORMAT, elementClass, caps);
    GRefPtr<GstElement> result(makeGStreamerElement(elementClass, encoderName.ascii().data()));
    if (!result && g_strcmp0(elementClass, "identity")) {
        GST_WARNING_OBJECT(bin, "Couldn't create %s, there might be problems processing some MSE streams. "
            "Continue at your own risk and consider adding %s to your build.", elementClass, elementClass);
        result = makeGStreamerElement("identity", encoderName.ascii().data());
    }
    return result;
}

std::pair<AppendPipeline::CreateTrackResult, AppendPipeline::Track*> AppendPipeline::tryCreateTrackFromPad(GstPad* demuxerSrcPad)
{
    ASSERT(isMainThread());
    ASSERT(!m_hasReceivedFirstInitializationSegment);
    GST_DEBUG_OBJECT(pipeline(), "Creating Track object for pad %" GST_PTR_FORMAT, demuxerSrcPad);

    const String& type = m_sourceBufferPrivate.type().containerType();
    if (type.endsWith("webm"_s))
        gst_pad_add_probe(demuxerSrcPad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, matroskademuxForceSegmentStartToEqualZero, nullptr, nullptr);

    auto [parsedCaps, streamType, presentationSize] = parseDemuxerSrcPadCaps(adoptGRef(gst_pad_get_current_caps(demuxerSrcPad)).get());
    GST_DEBUG_OBJECT(pipeline(), "Demuxer src pad caps: %" GST_PTR_FORMAT, parsedCaps.get());

    if (streamType == StreamType::Invalid) {
        GST_WARNING_OBJECT(pipeline(), "Unsupported track codec: %" GST_PTR_FORMAT, parsedCaps.get());
        // 3.5.7 Initialization Segment Received
        // 5.1. If the initialization segment contains tracks with codecs the user agent does not support, then run the
        // append error algorithm and abort these steps.
        return { CreateTrackResult::AppendParsingFailed, nullptr };
    }
    if (streamType == StreamType::Unknown) {
        GST_WARNING_OBJECT(pipeline(), "Pad '%s' with parsed caps %" GST_PTR_FORMAT " has an unknown type, will be connected to a black hole probe.", GST_PAD_NAME(demuxerSrcPad), parsedCaps.get());
        gst_pad_add_probe(demuxerSrcPad, GST_PAD_PROBE_TYPE_BUFFER, reinterpret_cast<GstPadProbeCallback>(appendPipelineDemuxerBlackHolePadProbe), nullptr, nullptr);
        return { CreateTrackResult::TrackIgnored, nullptr };
    }

    size_t newTrackIndex = m_tracks.size();
    size_t webkitTrackIndex = std::count_if(m_tracks.begin(), m_tracks.end(), [streamType](std::unique_ptr<Track>& other) -> bool {
        return other->streamType == streamType;
    });

    TrackID preferredTrackId = getStreamIdFromPad(demuxerSrcPad).value_or((static_cast<TrackID>(newTrackIndex)));
    auto assignedTrackId = m_sourceBufferPrivate.tryRegisterTrackId(preferredTrackId);
    TrackID trackId = assignedTrackId.value_or(preferredTrackId);

    if (!assignedTrackId)
        GST_WARNING_OBJECT(pipeline(), "Failed to register track ID %" PRIu64 ", this could cause ID collisions", preferredTrackId);

    GST_DEBUG_OBJECT(pipeline(), "Creating new AppendPipeline::Track with type %s, index %" PRIu64" and id '%" PRIu64 "'", streamTypeToString(streamType), webkitTrackIndex, trackId);
    m_tracks.append(makeUnique<Track>(trackId, streamType, parsedCaps, presentationSize));
    Track& track = *m_tracks.at(newTrackIndex);

    track.initializeElements(this, GST_BIN(m_pipeline.get()));
    track.webKitTrack = makeWebKitTrack(track, webkitTrackIndex, trackId);
    hookTrackEvents(track);
    return { CreateTrackResult::TrackCreated, &track };
}

bool AppendPipeline::recycleTrackForPad(GstPad* demuxerSrcPad)
{
    ASSERT(isMainThread());
    ASSERT(m_hasReceivedFirstInitializationSegment);
    auto trackId = AtomString(unsafeSpan8(GST_PAD_NAME(demuxerSrcPad)));
    auto [parsedCaps, streamType, presentationSize] = parseDemuxerSrcPadCaps(adoptGRef(gst_pad_get_current_caps(demuxerSrcPad)).get());

    GST_DEBUG_OBJECT(demuxerSrcPad, "Caps: %" GST_PTR_FORMAT, parsedCaps.get());

    // Try to find a matching pre-existing track. Ideally, tracks should be matched by track ID, but matching by type
    // is provided as a fallback -- which will be used, since we don't have a way to fetch those from GStreamer at the moment.
    Track* matchingTrack = nullptr;
    for (std::unique_ptr<Track>& track : m_tracks) {
        if (track->streamType != streamType)
            continue;
        matchingTrack = &*track;
    }

    if (!matchingTrack) {
        // Invalid configuration.
        GST_WARNING_OBJECT(pipeline(), "Couldn't find a matching pre-existing track for pad '%s' with parsed caps %" GST_PTR_FORMAT
            " on non-first initialization segment, will be connected to a black hole probe.", GST_PAD_NAME(demuxerSrcPad), parsedCaps.get());
        gst_pad_add_probe(demuxerSrcPad, GST_PAD_PROBE_TYPE_BUFFER, reinterpret_cast<GstPadProbeCallback>(appendPipelineDemuxerBlackHolePadProbe), nullptr, nullptr);
        return false;
    }

    // The https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/4535 merge request in qtdemux is causing EOS on
    // a "to be removed" stream before the no-more-pads message is triggered. That message makes AppendPipeline realize that
    // the stream is actually going to be removed. AppendPipeline may therefore be trying to reuse a former EOSed parser and
    // appsink elements from the old terminated stream, so "restarting" them (by stopping them now and restarting them at the
    // end) is required.
    // Note that what happens before qtdemux is irrelevant. Qtdemux itself is never receiving EOS in this case.
    if (matchingTrack->parser)
        gst_element_set_state(matchingTrack->parser.get(), GST_STATE_NULL);
    gst_element_set_state(matchingTrack->appsink.get(), GST_STATE_NULL);

    GRefPtr<GstCaps> matchingTrackCaps = adoptGRef(gst_pad_get_current_caps(matchingTrack->entryPad.get()));
    if (!matchingTrack->isLinked() && (!matchingTrackCaps || gst_caps_can_intersect(parsedCaps.get(), matchingTrackCaps.get())))
        linkPadWithTrack(demuxerSrcPad, *matchingTrack);
    else {
        // Unlink from old track and link to new track.
        auto peer = adoptGRef(gst_pad_get_peer(matchingTrack->entryPad.get()));
        if (peer.get() != demuxerSrcPad) {
            if (peer) {
                GST_DEBUG_OBJECT(peer.get(), "Unlinking from track %" PRIu64 "", matchingTrack->trackId);
                gst_pad_unlink(peer.get(), matchingTrack->entryPad.get());
            }

            const String& type = m_sourceBufferPrivate.type().containerType();
            if (type.endsWith("webm"_s))
                gst_pad_add_probe(demuxerSrcPad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, matroskademuxForceSegmentStartToEqualZero, nullptr, nullptr);

            matchingTrack->emplaceOptionalEncoderForFormat(GST_BIN_CAST(m_pipeline.get()), parsedCaps);
            matchingTrack->emplaceOptionalParserForFormat(GST_BIN_CAST(m_pipeline.get()), parsedCaps);
            linkPadWithTrack(demuxerSrcPad, *matchingTrack);
            matchingTrack->caps = WTFMove(parsedCaps);
            matchingTrack->presentationSize = presentationSize;
        } else
            GST_DEBUG_OBJECT(pipeline(), "%" PRIu64 " track pads match, nothing to re-link", matchingTrack->trackId);

    }

    gst_element_set_state(matchingTrack->appsink.get(), GST_STATE_PLAYING);
    if (matchingTrack->parser)
        gst_element_set_state(matchingTrack->parser.get(), GST_STATE_PLAYING);

    return true;
}

void AppendPipeline::linkPadWithTrack(GstPad* demuxerSrcPad, Track& track)
{
    GST_DEBUG_OBJECT(demuxerSrcPad, "Linking to track %" PRIu64 "", track.trackId);
    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, "append-pipeline-before-link");
    ASSERT(!GST_PAD_IS_LINKED(track.entryPad.get()));
    gst_pad_link(demuxerSrcPad, track.entryPad.get());
    ASSERT(GST_PAD_IS_LINKED(track.entryPad.get()));
    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, "append-pipeline-after-link");
}

Ref<WebCore::TrackPrivateBase> AppendPipeline::makeWebKitTrack(Track& appendPipelineTrack, int trackIndex, TrackID trackId)
{
    RefPtr<WebCore::TrackPrivateBase> track;
    TrackPrivateBaseGStreamer* gstreamerTrack = nullptr;
    // FIXME: AudioTrackPrivateGStreamer etc. should probably use pads of the playback pipeline rather than the append pipeline.
    GRefPtr<GstPad> pad(appendPipelineTrack.appsinkPad);
    switch (appendPipelineTrack.streamType) {
    case StreamType::Audio: {
        auto specificTrack = AudioTrackPrivateGStreamer::create(m_playerPrivate, trackIndex, WTFMove(pad), trackId);
        gstreamerTrack = specificTrack.ptr();
        track = static_cast<TrackPrivateBase*>(specificTrack.ptr());
        break;
    }
    case StreamType::Video: {
        auto specificTrack = VideoTrackPrivateGStreamer::create(m_playerPrivate, trackIndex, WTFMove(pad), trackId);
        gstreamerTrack = specificTrack.ptr();
        track = static_cast<TrackPrivateBase*>(specificTrack.ptr());
        break;
    }
    case StreamType::Text: {
        auto specificTrack = InbandTextTrackPrivateGStreamer::create(trackIndex, WTFMove(pad), trackId);
        gstreamerTrack = specificTrack.ptr();
        track = static_cast<TrackPrivateBase*>(specificTrack.ptr());
        break;
    }
    default:
        ASSERT_NOT_REACHED();
    }
    ASSERT(appendPipelineTrack.finalCaps.get());
    gstreamerTrack->setInitialCaps(appendPipelineTrack.finalCaps.get());
    return track.releaseNonNull();
}

void AppendPipeline::Track::emplaceOptionalParserForFormat(GstBin* bin, const GRefPtr<GstCaps>& newCaps)
{
    // Some audio files unhelpfully omit the duration of frames in the container. We need to parse
    // the contained audio streams in order to know the duration of the frames.
    // This is known to be an issue with YouTube WebM files containing Opus audio as of YTTV2018.
    // If no parser is needed, a GstIdentity element will be created instead.

    if (parser) {
        ASSERT(caps);
        // When switching from encrypted to unencrypted content the caps can change and we need to replace the parser.
        if (gstStructureGetName(gst_caps_get_structure(caps.get(), 0)) == gstStructureGetName(gst_caps_get_structure(newCaps.get(), 0))) {
            GST_TRACE_OBJECT(bin, "caps are compatible, bailing out");
            return;
        }

        GST_TRACE_OBJECT(bin, "caps are not compatible, replacing parser");
        auto locker = GstStateLocker(bin);
        gst_element_unlink(parser.get(), encoder.get());
        gst_element_set_state(parser.get(), GST_STATE_NULL);
        gst_bin_remove(bin, parser.get());
    }

    auto parserName = makeString("parser"_s, unsafeSpan8(streamTypeToStringLower(streamType)), "_"_s, trackId);
    parser = createOptionalParserForFormat(bin, parserName, newCaps.get());
    gst_bin_add(bin, parser.get());
    gst_element_sync_state_with_parent(parser.get());
    gst_element_link(parser.get(), encoder.get());
    ASSERT(GST_PAD_IS_LINKED(encoderPad.get()));
    entryPad = adoptGRef(gst_element_get_static_pad(parser.get(), "sink"));
}

void AppendPipeline::Track::emplaceOptionalEncoderForFormat(GstBin* bin, const GRefPtr<GstCaps>& newCaps)
{
    auto encoderName = makeString("encoder"_s, unsafeSpan8(streamTypeToStringLower(streamType)), "_"_s, trackId);
    encoder = createOptionalEncoderForFormat(bin, encoderName, newCaps.get());
    gst_bin_add(bin, encoder.get());
    gst_element_sync_state_with_parent(encoder.get());
    gst_element_link(encoder.get(), appsink.get());
    ASSERT(GST_PAD_IS_LINKED(appsinkPad.get()));
    encoderPad = adoptGRef(gst_element_get_static_pad(encoder.get(), "sink"));

    if (streamType == StreamType::Text)
        finalCaps = gst_caps_new_empty_simple("application/x-subtitle-vtt");
    else
        finalCaps = newCaps;

    GST_INFO_OBJECT(bin, "final caps for track %" PRIu64 ": %" GST_PTR_FORMAT, trackId, finalCaps.get());
}

void AppendPipeline::Track::initializeElements(AppendPipeline* appendPipeline, GstBin* bin)
{
    appsink = makeGStreamerElement("appsink", nullptr);
    gst_app_sink_set_emit_signals(GST_APP_SINK(appsink.get()), TRUE);
    gst_base_sink_set_sync(GST_BASE_SINK(appsink.get()), FALSE);
    gst_base_sink_set_async_enabled(GST_BASE_SINK(appsink.get()), FALSE); // No prerolls, no async state changes.
    gst_base_sink_set_drop_out_of_segment(GST_BASE_SINK(appsink.get()), FALSE);
    gst_base_sink_set_last_sample_enabled(GST_BASE_SINK(appsink.get()), FALSE);

    gst_bin_add(GST_BIN(appendPipeline->pipeline()), appsink.get());
    gst_element_sync_state_with_parent(appsink.get());
    appsinkPad = adoptGRef(gst_element_get_static_pad(appsink.get(), "sink"));

#if !LOG_DISABLED
    appsinkDataEnteringPadProbeInformation.appendPipeline = appendPipeline;
    appsinkDataEnteringPadProbeInformation.description = "appsink data entering";
    appsinkDataEnteringPadProbeInformation.probeId = gst_pad_add_probe(appsinkPad.get(), GST_PAD_PROBE_TYPE_BUFFER, reinterpret_cast<GstPadProbeCallback>(appendPipelinePadProbeDebugInformation), &appsinkDataEnteringPadProbeInformation, nullptr);
#endif

#if ENABLE(ENCRYPTED_MEDIA)
    appsinkPadEventProbeInformation.appendPipeline = appendPipeline;
    appsinkPadEventProbeInformation.description = "appsink event probe";
    appsinkPadEventProbeInformation.probeId = gst_pad_add_probe(appsinkPad.get(), GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, reinterpret_cast<GstPadProbeCallback>(appendPipelineAppsinkPadEventProbe), &appsinkPadEventProbeInformation, nullptr);
#endif

    emplaceOptionalEncoderForFormat(bin, caps);
    emplaceOptionalParserForFormat(bin, caps);
}

void AppendPipeline::hookTrackEvents(Track& track)
{
    g_signal_connect(track.appsink.get(), "new-sample", G_CALLBACK(+[](GstElement* appsink, AppendPipeline* appendPipeline) -> GstFlowReturn {
        appendPipeline->handleAppsinkNewSampleFromStreamingThread(appsink);
        return GST_FLOW_OK;
    }), this);

    struct Closure {
    public:

        Closure(AppendPipeline& appendPipeline, Track& track)
            : appendPipeline(appendPipeline)
            , track(track)
        { }
        static void destruct(void* closure, GClosure*) { delete static_cast<Closure*>(closure); }

        AppendPipeline& appendPipeline;
        Track& track;
    };

    g_signal_connect_data(track.appsinkPad.get(), "notify::caps", G_CALLBACK(+[](GObject*, GParamSpec*, Closure* closure) {
        AppendPipeline& appendPipeline = closure->appendPipeline;
        Track& track = closure->track;
        if (isMainThread()) {
            // When changing the pipeline state down to READY the demuxer is unlinked and this triggers a caps notification
            // because the appsink loses its previously negotiated caps. We are not interested in these unnegotiated caps.
#ifndef NDEBUG
            GRefPtr<GstPad> pad = adoptGRef(gst_element_get_static_pad(track.appsink.get(), "sink"));
            GRefPtr<GstCaps> caps = adoptGRef(gst_pad_get_current_caps(pad.get()));
            ASSERT(!caps);
#endif
            return;
        }

        // The streaming thread has just received a new caps and is about to let samples using the
        // new caps flow. Let's block it until the main thread has consumed the samples with the old
        // caps and has processed the caps change.
        appendPipeline.m_taskQueue.enqueueTaskAndWait<AbortableTaskQueue::Void>([&appendPipeline, &track]() {
            appendPipeline.appsinkCapsChanged(track);
            return AbortableTaskQueue::Void();
        });
    }), new Closure { *this, track }, Closure::destruct, static_cast<GConnectFlags>(0));
}

#ifndef GST_DISABLE_GST_DEBUG
const char* AppendPipeline::streamTypeToString(StreamType streamType)
{
    switch (streamType) {
    case StreamType::Audio:
        return "Audio";
    case StreamType::Video:
        return "Video";
    case StreamType::Text:
        return "Text";
    case StreamType::Invalid:
        return "Invalid";
    case StreamType::Unknown:
        return "Unknown";
    default:
        return "(Unsupported stream type)";
    }
}
#endif

const char* AppendPipeline::streamTypeToStringLower(StreamType streamType)
{
    switch (streamType) {
    case StreamType::Audio:
        return "audio";
    case StreamType::Video:
        return "video";
    case StreamType::Text:
        return "text";
    case StreamType::Invalid:
        return "invalid";
    case StreamType::Unknown:
        return "unknown";
    default:
        return "(unsupported_stream_type)";
    }
}

#if !LOG_DISABLED
static GstPadProbeReturn appendPipelinePadProbeDebugInformation(GstPad* pad, GstPadProbeInfo* info, struct PadProbeInformation* padProbeInformation)
{
    ASSERT(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER);
    GstBuffer* buffer = GST_PAD_PROBE_INFO_BUFFER(info);
    GST_TRACE_OBJECT(pad, "%s: buffer of size %" G_GSIZE_FORMAT " going through", padProbeInformation->description, gst_buffer_get_size(buffer));
    return GST_PAD_PROBE_OK;
}
#endif

#if ENABLE(ENCRYPTED_MEDIA)
static GstPadProbeReturn appendPipelineAppsinkPadEventProbe([[maybe_unused]] GstPad* pad, GstPadProbeInfo* info, struct PadProbeInformation *padProbeInformation)
{
    ASSERT(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM);
    GstEvent* event = gst_pad_probe_info_get_event(info);
    GST_DEBUG_OBJECT(pad, "Handling event %s on append pipeline appsinkPad", GST_EVENT_TYPE_NAME(event));
    AppendPipeline* appendPipeline = padProbeInformation->appendPipeline;

    switch (GST_EVENT_TYPE(event)) {
    case GST_EVENT_PROTECTION:
        if (appendPipeline && appendPipeline->playerPrivate())
            appendPipeline->playerPrivate()->handleProtectionEvent(event);
        return GST_PAD_PROBE_DROP;
    default:
        break;
    }

    return GST_PAD_PROBE_OK;
}
#endif

static GstPadProbeReturn appendPipelineDemuxerBlackHolePadProbe([[maybe_unused]] GstPad* pad, [[maybe_unused]] GstPadProbeInfo* info, gpointer)
{
    ASSERT(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER);
#ifndef GST_DISABLE_GST_DEBUG
    GstBuffer* buffer = GST_PAD_PROBE_INFO_BUFFER(info);
    GST_TRACE_OBJECT(pad, "buffer of size %" G_GSIZE_FORMAT " ignored", gst_buffer_get_size(buffer));
#endif
    return GST_PAD_PROBE_DROP;
}

static GstPadProbeReturn matroskademuxForceSegmentStartToEqualZero(GstPad*, GstPadProbeInfo* info, void*)
{
    // matroskademux sets GstSegment.start to the PTS of the first frame.
    //
    // This way in the unlikely case a user made a .mkv or .webm file where a certain portion of the movie is skipped
    // (e.g. by concatenating a MSE initialization segment with any MSE media segment other than the first) and opened
    // it with a regular player, playback would start immediately. GstSegment.duration is not modified in any case.
    //
    // Leaving the usefulness of that feature aside, the fact that it uses GstSegment.start is problematic for MSE.
    // In MSE is not unusual to process unordered MSE media segments. In this case, a frame may have
    // PTS <<< GstSegment.start and be discarded by downstream. This happens for instance in elements derived from
    // audiobasefilter, such as opusparse.
    //
    // This probe remedies the problem by setting GstSegment.start to 0 in all cases, not only when the PTS of the first
    // frame is zero.

    ASSERT(info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM);
    GstEvent* event = static_cast<GstEvent*>(info->data);
    if (event->type == GST_EVENT_SEGMENT) {
        GstSegment segment;
        gst_event_copy_segment(event, &segment);

        segment.start = 0;

        GRefPtr<GstEvent> newEvent = adoptGRef(gst_event_new_segment(&segment));
        gst_event_replace(reinterpret_cast<GstEvent**>(&info->data), newEvent.get());
    }
    return GST_PAD_PROBE_OK;
}

#undef GST_CAT_DEFAULT

} // namespace WebCore.

#endif // ENABLE(VIDEO) && USE(GSTREAMER) && ENABLE(MEDIA_SOURCE)