#include "cast_media.h"
extern "C" {
#include <libavcodec/avcodec.h>
#include <libavcodec/bsf.h>
#include <libavformat/avformat.h>
#include <libavutil/opt.h>
}
#include <algorithm>
#include <atomic>
#include <cassert>
#include <cctype>
#include <cstdlib>
#include <chrono>
#include <condition_variable>
#include <cmath>
#include <cstring>
#include <ctime>
#include <filesystem>
#include <fstream>
#include <future>
#include <iostream>
#include <iterator>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <poll.h>
#include <spawn.h>
#include <sys/resource.h>
#include <sys/wait.h>

extern char **environ;

// Encoded H.264 output cannot identify its input decoder; observe received
// source frames only in the check build, before CPU-readable mapping.
#ifdef CAST_MEDIA_CHECK
extern "C" void cast_media_check_decode_counts(uint64_t *, uint64_t *);
#endif
static std::pair<uint64_t, uint64_t> decodeCounts() {
    uint64_t hardware = 0, software = 0;
#ifdef CAST_MEDIA_CHECK
    cast_media_check_decode_counts(&hardware, &software);
#endif
    return {hardware, software};
}
static std::pair<uint64_t, uint64_t> decodeUsage(std::pair<uint64_t, uint64_t> before) {
    const auto after = decodeCounts();
    const std::pair counts{after.first - before.first, after.second - before.second};
#ifdef CAST_MEDIA_CHECK
    assert(counts.first + counts.second > 0);
#ifdef TEREBI_SOFTWARE_DECODE
    assert(counts.first == 0);
#endif
    std::cout << " decoder_videotoolbox_frames=" << counts.first << " decoder_software_frames=" << counts.second;
#else
    std::cout << " decoder_frames=unavailable (compile with CAST_MEDIA_CHECK)";
#endif
    return counts;
}

static void ok(int result) { assert(result >= 0); }
static std::vector<uint8_t> readFile(const std::string &path) {
    std::ifstream file(path, std::ios::binary); assert(file.good());
    return {std::istreambuf_iterator<char>(file), std::istreambuf_iterator<char>()};
}
static void writeFile(const std::string &path, const std::vector<uint8_t> &bytes) {
    std::ofstream file(path, std::ios::binary); file.write(reinterpret_cast<const char *>(bytes.data()), bytes.size()); assert(file.good());
}
// Real source A/V built with the same bundled encoders; no PATH ffmpeg or fixtures.
static void fixture(const std::string &path, int seconds = 90, int width = 96, int height = 64, int bitrate = 300000, bool compatible = false) {
    AVFormatContext *format = nullptr; ok(avformat_alloc_output_context2(&format, nullptr, "mpegts", path.c_str()));
    AVCodecContext *video = avcodec_alloc_context3(avcodec_find_encoder_by_name("h264_videotoolbox"));
    AVCodecContext *audio = avcodec_alloc_context3(avcodec_find_encoder(AV_CODEC_ID_AAC)); assert(video && audio);
    video->width = width; video->height = height; video->pix_fmt = AV_PIX_FMT_NV12;
    video->time_base = {1, 10}; video->framerate = {10, 1}; video->bit_rate = bitrate; video->gop_size = 70; video->max_b_frames = 0;
    video->sample_aspect_ratio = {4, 3};
    av_opt_set(video->priv_data, "allow_sw", "1", 0); av_opt_set(video->priv_data, "realtime", "1", 0);
    if (compatible) {
        ok(av_opt_set(video->priv_data, "profile", "main", 0));
        ok(av_opt_set(video->priv_data, "level", "4.1", 0));
    }
    audio->sample_rate = 48000; audio->sample_fmt = AV_SAMPLE_FMT_FLTP; audio->time_base = {1, 48000}; audio->bit_rate = 128000;
    av_channel_layout_default(&audio->ch_layout, 2);
    ok(avcodec_open2(video, video->codec, nullptr)); ok(avcodec_open2(audio, audio->codec, nullptr));
    auto *vs = avformat_new_stream(format, nullptr), *as = avformat_new_stream(format, nullptr); assert(vs && as);
    vs->time_base = video->time_base; as->time_base = audio->time_base;
    ok(avcodec_parameters_from_context(vs->codecpar, video)); ok(avcodec_parameters_from_context(as->codecpar, audio));
    AVBSFContext *aspect = nullptr;
    ok(av_bsf_alloc(av_bsf_get_by_name("h264_metadata"), &aspect));
    ok(avcodec_parameters_copy(aspect->par_in, vs->codecpar));
    aspect->time_base_in = video->time_base;
    ok(av_opt_set(aspect->priv_data, "sample_aspect_ratio", "4/3", 0));
    ok(av_bsf_init(aspect));
    ok(avio_open(&format->pb, path.c_str(), AVIO_FLAG_WRITE)); ok(avformat_write_header(format, nullptr));
    auto encode = [&](AVCodecContext *codec, AVStream *stream, AVFrame *frame) {
        ok(avcodec_send_frame(codec, frame)); auto *packet = av_packet_alloc(); assert(packet);
        for (;;) {
            int result = avcodec_receive_packet(codec, packet);
            if (result == AVERROR(EAGAIN) || result == AVERROR_EOF) break;
            ok(result);
            if (codec == video) {
                ok(av_bsf_send_packet(aspect, packet));
                while (av_bsf_receive_packet(aspect, packet) >= 0) {
                    av_packet_rescale_ts(packet, codec->time_base, stream->time_base); packet->stream_index = stream->index;
                    ok(av_interleaved_write_frame(format, packet)); av_packet_unref(packet);
                }
            } else {
                av_packet_rescale_ts(packet, codec->time_base, stream->time_base); packet->stream_index = stream->index;
                ok(av_interleaved_write_frame(format, packet)); av_packet_unref(packet);
            }
        }
        av_packet_free(&packet);
    };
    int64_t vi = 0, ai = 0;
    while (vi < seconds * 10 || ai < seconds * 48000) {
        bool isVideo = vi < seconds * 10 && (ai >= seconds * 48000 || vi / 10.0 <= ai / 48000.0);
        auto *f = av_frame_alloc(); assert(f);
        if (isVideo) {
            f->width = width; f->height = height; f->format = video->pix_fmt; f->pts = vi++;
            ok(av_frame_get_buffer(f, 32));
            for (int y = 0; y < height; ++y) std::memset(f->data[0] + y * f->linesize[0], 40 + vi % 160, width);
            for (int y = 0; y < height / 2; ++y) std::memset(f->data[1] + y * f->linesize[1], 128, width);
            encode(video, vs, f);
        } else {
            f->sample_rate = 48000; f->format = audio->sample_fmt; f->nb_samples = audio->frame_size;
            ok(av_channel_layout_copy(&f->ch_layout, &audio->ch_layout)); f->pts = ai;
            ok(av_frame_get_buffer(f, 0));
            for (int channel = 0; channel < 2; ++channel) for (int n = 0; n < f->nb_samples; ++n)
                reinterpret_cast<float *>(f->data[channel])[n] = float(.1 * std::sin((ai + n) * 440 * 2 * 3.141592653589793 / 48000));
            ai += f->nb_samples; encode(audio, as, f);
        }
        av_frame_free(&f);
    }
    encode(video, vs, nullptr); encode(audio, as, nullptr); ok(av_write_trailer(format));
    av_bsf_free(&aspect);
    avio_closep(&format->pb); avformat_free_context(format); avcodec_free_context(&video); avcodec_free_context(&audio);
}
// Indexed split MP4 is the resolved YouTube VOD shape. As with check-fixtures,
// the check requires host ffmpeg for MP4 muxing; production uses only bundled libraries.
static void splitFixture(const std::string &source, const std::string &path, AVMediaType type) {
    const char *arguments[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
        "-i", source.c_str(), "-map", type == AVMEDIA_TYPE_VIDEO ? "0:v:0" : "0:a:0",
        "-c", "copy", "-movflags", "+faststart", path.c_str(), nullptr};
    pid_t child = 0;
    int result = posix_spawnp(&child, arguments[0], nullptr, nullptr,
        const_cast<char *const *>(arguments), environ);
    if (result != 0) std::cerr << "cast-media split fixture requires host ffmpeg (same as make check-fixtures)\n";
    assert(result == 0);
    int status = 0; assert(waitpid(child, &status, 0) == child);
    assert(WIFEXITED(status) && WEXITSTATUS(status) == 0);
}
static std::map<std::string, std::vector<uint8_t>> hlsFixture(const std::string &source, const std::string &unusedSource, const std::filesystem::path &directory) {
    std::filesystem::create_directory(directory);
    for (const auto &variant : {std::pair{"selected", source}, std::pair{"unused", unusedSource}}) {
        const auto playlist = (directory / (std::string(variant.first) + ".m3u8")).string();
        const auto segments = (directory / (std::string(variant.first) + "-%03d.ts")).string();
        const char *arguments[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
            "-i", variant.second.c_str(), "-map", "0", "-c", "copy", "-hls_time", "6",
            "-hls_list_size", "0", "-hls_segment_filename", segments.c_str(), playlist.c_str(), nullptr};
        pid_t child = 0;
        assert(posix_spawnp(&child, arguments[0], nullptr, nullptr, const_cast<char *const *>(arguments), environ) == 0);
        int status = 0; assert(waitpid(child, &status, 0) == child);
        assert(WIFEXITED(status) && WEXITSTATUS(status) == 0);
    }
    std::map<std::string, std::vector<uint8_t>> resources;
    for (const auto &entry : std::filesystem::directory_iterator(directory))
        resources["/hls/" + entry.path().filename().string()] = readFile(entry.path().string());
    const std::string master = "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=400000,RESOLUTION=96x64,CODECS=\"avc1.64000a,mp4a.40.2\"\nselected.m3u8\n"
        "#EXT-X-STREAM-INF:BANDWIDTH=200000,RESOLUTION=48x32,CODECS=\"avc1.64000a,mp4a.40.2\"\nunused.m3u8\n";
    resources["/hls/master.m3u8"] = {master.begin(), master.end()};
    return resources;
}
static std::vector<uint8_t> wav() {
    std::vector<uint8_t> bytes(44 + 90 * 48000 * 2);
    auto put = [&](size_t at, uint32_t value, int length) { for (int i = 0; i < length; ++i) bytes[at + i] = uint8_t(value >> (8 * i)); };
    std::memcpy(bytes.data(), "RIFF", 4); put(4, uint32_t(bytes.size() - 8), 4); std::memcpy(bytes.data() + 8, "WAVEfmt ", 8);
    put(16, 16, 4); put(20, 1, 2); put(22, 1, 2); put(24, 48000, 4); put(28, 96000, 4); put(32, 2, 2); put(34, 16, 2);
    std::memcpy(bytes.data() + 36, "data", 4); put(40, uint32_t(bytes.size() - 44), 4);
    for (size_t i = 0; i < 90 * 48000; ++i) put(44 + i * 2, uint16_t(int16_t(4000 * std::sin(i * 440 * 2 * 3.141592653589793 / 48000))), 2);
    return bytes;
}
struct HTTP {
    int status = 0; std::string headers; std::vector<uint8_t> bytes;
};
static HTTP request(const std::string &url, const std::string &method = "GET", const std::string &headers = "", bool allowClosed = false, std::promise<void> *sent = nullptr) {
    assert(url.starts_with("http://127.0.0.1:")); size_t slash = url.find('/', 7), colon = url.find(':', 7);
    sockaddr_in address{}; address.sin_family = AF_INET; address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); address.sin_port = htons(std::stoi(url.substr(colon + 1, slash - colon - 1)));
    int socket = ::socket(AF_INET, SOCK_STREAM, 0); assert(socket >= 0);
    timeval timeout{60, 0}; setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
    assert(connect(socket, reinterpret_cast<sockaddr *>(&address), sizeof(address)) == 0);
    std::string text = method + " " + url.substr(slash) + " HTTP/1.1\r\nHost: 127.0.0.1\r\n" + headers + "Connection: close\r\n\r\n";
    assert(send(socket, text.data(), text.size(), 0) == ssize_t(text.size()));
    if (sent) sent->set_value();
    std::vector<uint8_t> bytes; uint8_t buffer[65536];
    for (;;) {
        ssize_t n = recv(socket, buffer, sizeof(buffer), 0);
        if (n < 0 && allowClosed) break;
        assert(n >= 0); if (!n) break;
        bytes.insert(bytes.end(), buffer, buffer + n);
    }
    close(socket); std::string raw(bytes.begin(), bytes.end()); size_t end = raw.find("\r\n\r\n");
    if (allowClosed && end == std::string::npos) return {};
    assert(end != std::string::npos);
    return {std::stoi(raw.substr(9, 3)), raw.substr(0, end + 4), {bytes.begin() + end + 4, bytes.end()}};
}
// A tiny real upstream tests resource-specific private headers and byte-range seek.
struct Upstream {
    int listener; uint16_t port; std::thread threads[4]; std::atomic<bool> stopped{false};
    std::atomic<int> foreignRequests{0}, videoRequests{0}, audioRequests{0};
    std::atomic<int> videoStarts{0}, audioStarts{0};
    std::map<std::string, std::vector<uint8_t>> resources;
    std::mutex mutex;
    std::condition_variable gate;
    std::map<std::string, int> mediaRequests;
    bool gated = false;
    int held = 0, released = 0;
    std::map<std::string, int> counts() {
        std::lock_guard lock(mutex);
        return mediaRequests;
    }
    std::string foreign;
    explicit Upstream(std::vector<uint8_t> video, std::vector<uint8_t> audio) {
        resources["/source.ts"] = std::move(video); resources["/audio.wav"] = std::move(audio);
        resources["/public.ts"] = resources["/source.ts"];
        listener = socket(AF_INET, SOCK_STREAM, 0); assert(listener >= 0); int one = 1; setsockopt(listener, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
        sockaddr_in address{}; address.sin_family = AF_INET; address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        assert(bind(listener, reinterpret_cast<sockaddr *>(&address), sizeof(address)) == 0); assert(listen(listener, 16) == 0);
        assert(fcntl(listener, F_SETFL, O_NONBLOCK) == 0);
        socklen_t length = sizeof(address); assert(getsockname(listener, reinterpret_cast<sockaddr *>(&address), &length) == 0); port = ntohs(address.sin_port);
    }
    std::string url(std::string path) const { return "http://127.0.0.1:" + std::to_string(port) + path; }
    void start() { for (auto &thread : threads) thread = std::thread([&] {
        while (!stopped) {
            pollfd ready{listener, POLLIN, 0};
            if (poll(&ready, 1, 100) <= 0 || !(ready.revents & POLLIN)) continue;
            int client = accept(listener, nullptr, nullptr); if (client < 0) continue;
            assert(fcntl(client, F_SETFL, fcntl(client, F_GETFL) & ~O_NONBLOCK) == 0);
            int one = 1; setsockopt(client, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one)); timeval timeout{5, 0}; setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); setsockopt(client, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout));
            std::string request; char buffer[2048];
            while (request.find("\r\n\r\n") == std::string::npos && request.size() < 16384) { auto n = recv(client, buffer, sizeof(buffer), 0); if (n <= 0) break; request.append(buffer, size_t(n)); }
            size_t begin = request.find(' '), end = request.find(' ', begin + 1); std::string path = request.substr(begin + 1, end - begin - 1);
            int status = 200; std::string extra; std::vector<uint8_t> data;
            if (path == "/redirect") { status = 302; extra = "Location: " + foreign + "/source.ts\r\n"; }
            else if (path == "/cross.m3u8" || path == "/index.m3u8" || path == "/public-cross.m3u8") {
                std::string playlist = "#EXTM3U\n#EXT-X-TARGETDURATION:90\n#EXTINF:90,\n" +
                    (path == "/index.m3u8" ? url("") : foreign) +
                    (path == "/public-cross.m3u8" ? "/public.ts" : "/source.ts") + "\n#EXT-X-ENDLIST\n";
                data.assign(playlist.begin(), playlist.end());
            } else if (auto resource = resources.find(path); resource != resources.end()) {
                bool audio = path == "/audio.wav" || path == "/audio.m4a";
                bool publicMedia = path == "/public.ts";
                bool authorized = publicMedia || request.find(audio ? "X-Audio: audio-secret" : "X-Video: video-secret") != std::string::npos;
                bool wrong = request.find(publicMedia ? "secret" : (audio ? "video-secret" : "audio-secret")) != std::string::npos;
                if (!authorized || wrong) { ++foreignRequests; status = 403; }
                else {
                    if (audio) ++audioRequests; else ++videoRequests;
                    data = resource->second;
                    if (path.starts_with("/hls/") && path.ends_with(".ts") && request.starts_with("GET")) {
                        std::unique_lock lock(mutex);
                        ++mediaRequests[path];
                        if (gated) {
                            int ticket = ++held;
                            gate.notify_all();
                            gate.wait(lock, [&] { return released >= ticket || stopped; });
                        }
                    }
                }
            } else status = 404;
            size_t low = 0, high = data.size();
            auto range = request.find("Range: bytes=");
            if (range != std::string::npos && !data.empty()) {
                low = std::stoull(request.substr(range + 13)); auto dash = request.find('-', range);
                if (dash != std::string::npos && std::isdigit(request[dash + 1])) high = std::min(high, size_t(std::stoull(request.substr(dash + 1))) + 1);
                if (low >= high) { status = 416; low = high = 0; }
                else { status = 206; extra += "Content-Range: bytes " + std::to_string(low) + "-" + std::to_string(high - 1) + "/" + std::to_string(data.size()) + "\r\n"; }
            }
            if ((status == 200 || status == 206) && low == 0 && resources.contains(path)) {
                if (path == "/audio.wav" || path == "/audio.m4a") ++audioStarts; else ++videoStarts;
            }
            std::string header = "HTTP/1.1 " + std::to_string(status) + " Test\r\nConnection: close\r\nAccept-Ranges: bytes\r\n" + extra + "Content-Length: " + std::to_string(high - low) + "\r\n\r\n";
            send(client, header.data(), header.size(), 0);
            if (!request.starts_with("HEAD")) while (high > low) { auto n = send(client, data.data() + low, high - low, 0); if (n <= 0) break; low += size_t(n); }
            close(client);
        }
    }); }
    ~Upstream() {
        { std::lock_guard lock(mutex); stopped = true; gate.notify_all(); }
        shutdown(listener, SHUT_RDWR);
        for (auto &thread : threads) if (thread.joinable()) thread.join();
        close(listener);
    }
};
// Every media GET really waits at the upstream until this driver releases it.
// Deadlines are watchdogs, not media clocks; no elapsed-time sleeps drive the test.
template<typename Operation> static auto throughGate(Upstream &upstream, Operation operation) {
    bool done = false;
    auto task = std::async(std::launch::async, [&] {
        auto result = operation();
        { std::lock_guard lock(upstream.mutex); done = true; upstream.gate.notify_all(); }
        return result;
    });
    std::unique_lock lock(upstream.mutex);
    while (!done) {
        assert(upstream.gate.wait_for(lock, std::chrono::seconds(60), [&] { return done || upstream.held > upstream.released; }));
        upstream.released = upstream.held;
        upstream.gate.notify_all();
    }
    lock.unlock();
    return task.get();
}
static void playable(const HTTP &response, const std::string &path, bool video, double minimumTime, double maximumTime = INFINITY) {
    assert(response.status == 200 && response.bytes.size() > 188 && response.bytes[0] == 0x47); writeFile(path, response.bytes);
    AVFormatContext *format = nullptr; ok(avformat_open_input(&format, path.c_str(), nullptr, nullptr)); ok(avformat_find_stream_info(format, nullptr));
    int audioIndex = av_find_best_stream(format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0); assert(audioIndex >= 0);
    int videoIndex = av_find_best_stream(format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0); assert(!video || videoIndex >= 0);
    std::map<int, AVCodecContext *> codecs; std::map<int, int> frames;
    std::map<int, double> first;
    for (int index : {audioIndex, videoIndex}) if (index >= 0) {
        auto *stream = format->streams[index]; auto *codec = avcodec_alloc_context3(avcodec_find_decoder(stream->codecpar->codec_id)); assert(codec);
        ok(avcodec_parameters_to_context(codec, stream->codecpar)); ok(avcodec_open2(codec, codec->codec, nullptr)); codecs[index] = codec;
    }
    auto *packet = av_packet_alloc(); auto *frame = av_frame_alloc(); bool timestamp = false;
    while (av_read_frame(format, packet) >= 0) {
        auto found = codecs.find(packet->stream_index);
        if (found != codecs.end()) {
            if (packet->pts != AV_NOPTS_VALUE) {
                double time = packet->pts * av_q2d(format->streams[packet->stream_index]->time_base);
                assert(time >= minimumTime - .1 && time <= maximumTime + .1);
                if (!first.contains(packet->stream_index)) first[packet->stream_index] = time;
                timestamp = true;
            }
            ok(avcodec_send_packet(found->second, packet));
            while (avcodec_receive_frame(found->second, frame) >= 0) {
                if (packet->stream_index == videoIndex) {
                    if (std::isfinite(maximumTime)) assert(frame->height == 64);
                    auto sar = av_guess_sample_aspect_ratio(format, format->streams[videoIndex], frame);
                    if (sar.num <= 0 || sar.den <= 0) sar = {1, 1};
                    assert(sar.num > 0 && std::abs(double(frame->width) / frame->height * av_q2d(sar) - 2.0) < .01);
                }
                ++frames[packet->stream_index]; av_frame_unref(frame);
            }
        }
        av_packet_unref(packet);
    }
    assert(timestamp && frames[audioIndex] > 10 && (!video || frames[videoIndex] > 10));
    if (std::isfinite(maximumTime)) for (int index : {audioIndex, videoIndex}) if (index >= 0)
        assert(first.contains(index) && std::abs(first[index] - minimumTime) < .1);
    for (auto &[index, codec] : codecs) avcodec_free_context(&codec);
    av_packet_free(&packet); av_frame_free(&frame); avformat_close_input(&format);
}
static std::pair<double, double> audioSpan(const HTTP &response, const std::string &path) {
    assert(response.status == 200);
    writeFile(path, response.bytes);
    AVFormatContext *format = nullptr;
    ok(avformat_open_input(&format, path.c_str(), nullptr, nullptr));
    ok(avformat_find_stream_info(format, nullptr));
    int index = av_find_best_stream(format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0); assert(index >= 0);
    auto *codec = avcodec_alloc_context3(avcodec_find_decoder(format->streams[index]->codecpar->codec_id)); assert(codec);
    ok(avcodec_parameters_to_context(codec, format->streams[index]->codecpar)); ok(avcodec_open2(codec, codec->codec, nullptr));
    auto *packet = av_packet_alloc(); auto *frame = av_frame_alloc();
    double first = INFINITY, end = -INFINITY;
    while (av_read_frame(format, packet) >= 0) {
        if (packet->stream_index == index) {
            ok(avcodec_send_packet(codec, packet));
            while (avcodec_receive_frame(codec, frame) >= 0) {
                assert(frame->best_effort_timestamp != AV_NOPTS_VALUE);
                double position = frame->best_effort_timestamp * av_q2d(format->streams[index]->time_base);
                first = std::min(first, position);
                end = position + double(frame->nb_samples) / frame->sample_rate;
                av_frame_unref(frame);
            }
        }
        av_packet_unref(packet);
    }
    av_packet_free(&packet); av_frame_free(&frame); avcodec_free_context(&codec); avformat_close_input(&format);
    assert(std::isfinite(first) && end > first);
    return {first, end};
}
static CastMedia *start(CastMediaSource source, std::string &url, double &duration) {
    char error[512]{}, bytes[512]{}; auto *media = cast_media_create(&source, error); assert(media);
    assert(cast_media_start(media, "127.0.0.1", bytes, &duration, error)); url = bytes; return media;
}
static void seekWork(Upstream &upstream, const std::string &segment) {
    std::string video = upstream.url("/video.mp4"), audio = upstream.url("/audio.m4a"), url;
    double duration = 0;
    CastMediaSource source{video.c_str(), "X-Video: video-secret\r\n", audio.c_str(), "X-Audio: audio-secret\r\n", false, false, false, 90};
    auto *media = start(source, url, duration);
    std::string base = url.substr(0, url.rfind('/') + 1);
    int videoStarts = upstream.videoStarts, audioStarts = upstream.audioStarts;
    auto distant = request(base + "segment/10.ts");
    playable(distant, segment, true, 60);
    int videoRequests = upstream.videoRequests, audioRequests = upstream.audioRequests;
    auto repeated = request(base + "segment/10.ts");
    std::cout << "seek-work repeated upstream requests: video=" << int(upstream.videoRequests) - videoRequests
              << " audio=" << int(upstream.audioRequests) - audioRequests << std::endl;
    bool repeatedHasNoIO = upstream.videoRequests == videoRequests && upstream.audioRequests == audioRequests;
    bool repeatedIdentical = repeated.status == 200 && repeated.bytes == distant.bytes;
    auto adjacent = request(base + "segment/11.ts");
    playable(adjacent, segment, true, 66);
    auto before = audioSpan(distant, segment), after = audioSpan(adjacent, segment);
    assert(std::abs(after.first - before.second) < 1.0 / 48000);
    playable(request(base + "segment/2.ts"), segment, true, 12);
    playable(request(base + "segment/12.ts"), segment, true, 72);
    std::cout << "seek-work new origin-zero requests across distant/adjacent seeks: video="
              << int(upstream.videoStarts) - videoStarts << " audio=" << int(upstream.audioStarts) - audioStarts << std::endl;
    bool seeksDoNotReprobe = upstream.videoStarts == videoStarts && upstream.audioStarts == audioStarts;
    // A previous random segment remains independently replayable without upstream IO.
    videoRequests = upstream.videoRequests; audioRequests = upstream.audioRequests;
    bool replayIdentical = request(base + "segment/10.ts").bytes == distant.bytes;
    auto head = request(base + "segment/10.ts", "HEAD");
    auto range = request(base + "segment/10.ts", "GET", "Range: bytes=0-187\r\n");
    assert(head.status == 200 && head.bytes.empty());
    assert(range.status == 206 && range.bytes == std::vector<uint8_t>(distant.bytes.begin(), distant.bytes.begin() + 188));
    bool replayHasNoIO = upstream.videoRequests == videoRequests && upstream.audioRequests == audioRequests;
    cast_media_destroy(media);
    assert(seeksDoNotReprobe);
    assert(repeatedHasNoIO && repeatedIdentical && replayHasNoIO && replayIdentical);
}
static int mediaCount(const std::map<std::string, int> &counts, const std::string &prefix) {
    int total = 0;
    for (const auto &[path, count] : counts) if (path.starts_with(prefix)) total += count;
    return total;
}
static void retainedMedia(const char *stage, const std::map<std::string, int> &before, const std::map<std::string, int> &after) {
    for (const auto &[path, count] : after) {
        int previous = before.contains(path) ? before.at(path) : 0;
        int allowed = path.starts_with("/hls/unused-") ? previous : std::max(previous, 1);
        if (count > allowed)
            std::cerr << "hls-work retention failure: stage=" << stage << " media=" << path
                      << " previous=" << previous << " actual=" << count << " allowed=" << allowed << std::endl;
        if (path.starts_with("/hls/unused-")) assert(count == previous);
        else assert(count <= std::max(previous, 1));
    }
}
static void hlsWork(Upstream &upstream, const std::string &segment) {
    std::string hls = upstream.url("/hls/master.m3u8"), url;
    double duration = 0;
    CastMediaSource source{hls.c_str(), "X-Video: video-secret\r\n", nullptr, nullptr, false, true, false, 180};
    auto *media = throughGate(upstream, [&] { return start(source, url, duration); });
    assert(duration >= 179 && duration <= 182);
    const auto prepared = upstream.counts();
    assert(mediaCount(prepared, "/hls/selected-") > 0 && upstream.foreignRequests == 0);
    const std::string base = url.substr(0, url.rfind('/') + 1);
    auto fetch = [&](int id) {
        const auto before = upstream.counts();
        std::cout << "hls-work requesting segment=" << id << " position=" << id * 6 << std::endl;
        auto response = throughGate(upstream, [&] { return request(base + "segment/" + std::to_string(id) + ".ts"); });
        for (const auto &[path, count] : upstream.counts()) {
            int previous = before.contains(path) ? before.at(path) : 0;
            if (count != previous)
                std::cout << "hls-work segment=" << id << " media=" << path
                          << " previous=" << previous << " actual=" << count << std::endl;
        }
        // The relay's shared positive MPEG-TS clock origin is one second; media positions remain id * 6.
        playable(response, segment, true, id * 6 + 1, (id + 1) * 6 + 1);
        return response;
    };
    auto first = fetch(2), adjacent = fetch(3);
    auto firstAudio = audioSpan(first, segment), adjacentAudio = audioSpan(adjacent, segment);
    assert(std::abs(adjacentAudio.first - firstAudio.second) < 1.0 / 48000);
    fetch(1); // New output segment, but both A/V lanes must rewind retained source packets.
    assert(fetch(2).bytes == first.bytes);
    const auto early = upstream.counts();
    std::cout << "hls-work media GETs after preparation: selected="
              << mediaCount(early, "/hls/selected-") - mediaCount(prepared, "/hls/selected-")
              << " unused=" << mediaCount(early, "/hls/unused-") - mediaCount(prepared, "/hls/unused-") << std::endl;
    retainedMedia("prepared -> early 2,3,1,2", prepared, early);

    // Beyond the 60-second read-ahead window: this random seek requires real upstream IO.
    auto distant = fetch(24);
    const auto distantCounts = upstream.counts();
    assert(mediaCount(distantCounts, "/hls/selected-") > mediaCount(early, "/hls/selected-"));
    retainedMedia("early -> random 24", early, distantCounts);
    auto distantAdjacent = fetch(26);
    auto nearby = fetch(25); // Unrendered nearby output rewinds already-read source, not speculative history before a miss.
    auto nearbyAudio = audioSpan(nearby, segment), distantAdjacentAudio = audioSpan(distantAdjacent, segment);
    assert(std::abs(distantAdjacentAudio.first - nearbyAudio.second) < 1.0 / 48000);
    assert(fetch(24).bytes == distant.bytes);
    const auto nearbyCounts = upstream.counts();
    retainedMedia("random 24 -> 26,25,24", distantCounts, nearbyCounts);

    // Behind the 30-second retained window: a backward random miss must still decode at the target.
    auto backward = fetch(5);
    const auto backwardCounts = upstream.counts();
    assert(mediaCount(backwardCounts, "/hls/selected-") > mediaCount(nearbyCounts, "/hls/selected-"));
    assert(mediaCount(backwardCounts, "/hls/unused-") == mediaCount(prepared, "/hls/unused-"));
    assert(fetch(5).bytes == backward.bytes);
    retainedMedia("backward 5 -> replay 5", backwardCounts, upstream.counts());

    // Do not release the next media request. Cancel/destroy must finish while upstream is held.
    std::promise<void> sent;
    auto sentRequest = sent.get_future();
    auto blocked = std::async(std::launch::async, [&] { return request(base + "segment/20.ts", "GET", "", true, &sent); });
    assert(sentRequest.wait_for(std::chrono::seconds(10)) == std::future_status::ready);
    int released = 0;
    {
        std::unique_lock lock(upstream.mutex);
        assert(upstream.gate.wait_for(lock, std::chrono::seconds(10), [&] { return upstream.held > upstream.released; }));
        released = upstream.released;
    }
    assert(blocked.wait_for(std::chrono::seconds(0)) == std::future_status::timeout);
    auto destroyed = std::async(std::launch::async, [&] { cast_media_cancel(media); cast_media_destroy(media); });
    assert(destroyed.wait_for(std::chrono::seconds(3)) == std::future_status::ready);
    destroyed.get();
    assert(blocked.wait_for(std::chrono::seconds(3)) == std::future_status::ready);
    auto cancelled = blocked.get(); assert(cancelled.status == 0 || cancelled.status == 503);
    {
        std::lock_guard lock(upstream.mutex);
        assert(upstream.released == released && upstream.held > upstream.released);
        upstream.released = upstream.held;
        upstream.gate.notify_all();
    }
    assert(upstream.foreignRequests == 0);
}
struct CompressedSample {
    int64_t dts;
    std::vector<uint8_t> bytes;
    bool operator==(const CompressedSample &) const = default;
};
using Samples = std::map<std::pair<int, int64_t>, CompressedSample>;
// Compare actual AAC payloads and H.264 slices, excluding container-inserted
// AUD/parameter sets. Also decode each TS independently with strict errors.
static Samples samples(const std::string &path, double origin = NAN) {
    AVFormatContext *format = nullptr;
    ok(avformat_open_input(&format, path.c_str(), nullptr, nullptr)); ok(avformat_find_stream_info(format, nullptr));
    if (std::isnan(origin)) origin = double(format->start_time) / AV_TIME_BASE;
    Samples result;
    std::map<int, AVCodecContext *> decoders;
    std::map<int, int64_t> lastDTS;
    std::map<int, int> decoded;
    auto *packet = av_packet_alloc(); auto *frame = av_frame_alloc(); assert(packet && frame);
    while (av_read_frame(format, packet) >= 0) {
        auto *stream = format->streams[packet->stream_index];
        const bool video = stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO;
        assert(stream->codecpar->codec_id == (video ? AV_CODEC_ID_H264 : AV_CODEC_ID_AAC));
        assert(packet->pts != AV_NOPTS_VALUE && packet->dts != AV_NOPTS_VALUE);
        const int64_t pts = std::llround((packet->pts * av_q2d(stream->time_base) - origin) * 90000);
        const int64_t dts = std::llround((packet->dts * av_q2d(stream->time_base) - origin) * 90000);
        if (lastDTS.contains(video)) assert(dts > lastDTS[video]);
        lastDTS[video] = dts;
        std::vector<uint8_t> payload;
        if (video) {
            bool idr = false;
            for (int at = 0; at + 3 < packet->size;) {
                if (packet->data[at] || packet->data[at + 1] || packet->data[at + 2] != 1) { ++at; continue; }
                const int begin = at + 3, type = packet->data[begin] & 31;
                at = begin + 1;
                while (at + 3 < packet->size &&
                       (packet->data[at] || packet->data[at + 1] || packet->data[at + 2] != 1)) ++at;
                int end = at + 3 < packet->size ? at : packet->size;
                while (end > begin && packet->data[end - 1] == 0) --end;
                if (type == 1 || type == 5) payload.insert(payload.end(), packet->data + begin, packet->data + end);
                idr = idr || type == 5;
            }
            if (!decoders.contains(video)) assert((packet->flags & AV_PKT_FLAG_KEY) && idr);
            assert(!payload.empty());
        } else payload.assign(packet->data, packet->data + packet->size);
        assert(result.emplace(std::pair{int(video), pts}, CompressedSample{dts, std::move(payload)}).second);
        if (!decoders.contains(video)) {
            auto *decoder = avcodec_alloc_context3(avcodec_find_decoder(stream->codecpar->codec_id)); assert(decoder);
            ok(avcodec_parameters_to_context(decoder, stream->codecpar));
            decoder->err_recognition = AV_EF_EXPLODE;
            ok(avcodec_open2(decoder, decoder->codec, nullptr)); decoders[video] = decoder;
        }
        ok(avcodec_send_packet(decoders[video], packet));
        while (avcodec_receive_frame(decoders[video], frame) >= 0) { ++decoded[video]; av_frame_unref(frame); }
        av_packet_unref(packet);
    }
    for (auto &[type, decoder] : decoders) {
        ok(avcodec_send_packet(decoder, nullptr));
        while (avcodec_receive_frame(decoder, frame) >= 0) { ++decoded[type]; av_frame_unref(frame); }
        assert(decoded[type] == int(std::count_if(result.begin(), result.end(), [&](const auto &row) { return row.first.first == type; })));
        avcodec_free_context(&decoder);
    }
    assert(decoded[0] > 0 && decoded[1] > 0);
    av_packet_free(&packet); av_frame_free(&frame); avformat_close_input(&format);
    return result;
}
// Indexed, seven-second GOPs used to transcode at every six-second boundary.
static void remuxWork(const std::filesystem::path &directory) {
    const auto source = (directory / "remux.ts").string(), indexed = (directory / "remux.mp4").string();
    fixture(source, 28, 1280, 720, 3000000, true);
    // Keep audio with the indexed video so the check covers both packet lanes.
    const char *arguments[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
        "-i", source.c_str(), "-map", "0", "-c", "copy", "-movflags", "+faststart", indexed.c_str(), nullptr};
    pid_t child = 0;
    assert(posix_spawnp(&child, arguments[0], nullptr, nullptr, const_cast<char *const *>(arguments), environ) == 0);
    int status = 0; assert(waitpid(child, &status, 0) == child);
    assert(WIFEXITED(status) && WEXITSTATUS(status) == 0);
    std::string url; double duration = 0;
    auto began = std::chrono::steady_clock::now();
    auto *media = start({indexed.c_str(), "", nullptr, nullptr, true, false, false, 0}, url, duration);
    const auto prepared = std::chrono::steady_clock::now();
    auto manifest = request(url); assert(manifest.status == 200);
    std::string text(manifest.bytes.begin(), manifest.bytes.end()), base = url.substr(0, url.rfind('/') + 1);
    std::vector<double> boundaries{0};
    for (size_t at = 0; (at = text.find("#EXTINF:", at)) != std::string::npos; ++at)
        boundaries.push_back(boundaries.back() + std::stod(text.substr(at + 8)));
    size_t bytes = 0;
    std::vector<HTTP> outputs;
    for (size_t id = 0; id + 1 < boundaries.size(); ++id) {
        auto response = request(base + "segment/" + std::to_string(id) + ".ts");
        assert(response.status == 200);
        bytes += response.bytes.size(); outputs.push_back(std::move(response));
    }
    const auto finished = std::chrono::steady_clock::now();
    double longest = 0;
    for (size_t i = 1; i < boundaries.size(); ++i) longest = std::max(longest, boundaries[i] - boundaries[i - 1]);
    std::cout << "remux-work prepare_ms=" << std::chrono::duration<double, std::milli>(prepared - began).count()
              << " generate_ms=" << std::chrono::duration<double, std::milli>(finished - prepared).count()
              << " segments=" << outputs.size() << " bytes=" << bytes
              << " max_seconds=" << longest << std::endl;
    assert(request(base + "segment/" + std::to_string(outputs.size()) + ".ts").status == 404);
    assert(boundaries.size() == 5);
    for (size_t i = 0; i < boundaries.size(); ++i) assert(std::abs(boundaries[i] - i * 7.0) < .1);
    assert(text.find("#EXT-X-TARGETDURATION:8\n") != std::string::npos);
    Samples delivered;
    for (size_t id = 0; id < outputs.size(); ++id) {
        const auto path = (directory / "output.ts").string(); writeFile(path, outputs[id].bytes);
        auto rows = samples(path, 1);
        for (auto &[key, value] : rows) {
            assert(double(key.second) / 90000 >= boundaries[id] - .00002);
            assert(double(key.second) / 90000 < boundaries[id + 1] + .00002);
            assert(delivered.emplace(key, std::move(value)).second);
        }
    }
    assert(delivered == samples(source));
    cast_media_destroy(media);

    const auto video = (directory / "indexed-video.mp4").string(), audio = (directory / "indexed-audio.m4a").string();
    splitFixture(source, video, AVMEDIA_TYPE_VIDEO); splitFixture(source, audio, AVMEDIA_TYPE_AUDIO);
    Upstream upstream(readFile(indexed), wav());
    upstream.resources["/video.mp4"] = readFile(video); upstream.resources["/audio.m4a"] = readFile(audio);
    upstream.start();
    for (bool split : {false, true}) {
        const auto videoURL = upstream.url(split ? "/video.mp4" : "/source.ts"), audioURL = upstream.url("/audio.m4a");
        media = start({videoURL.c_str(), "X-Video: video-secret\r\n", split ? audioURL.c_str() : nullptr,
                       split ? "X-Audio: audio-secret\r\n" : nullptr, false, false, false, 0}, url, duration);
        base = url.substr(0, url.rfind('/') + 1);
        manifest = request(url); assert(manifest.status == 200);
        text.assign(manifest.bytes.begin(), manifest.bytes.end());
        std::vector<double> networkBoundaries{0};
        for (size_t at = 0; (at = text.find("#EXTINF:", at)) != std::string::npos; ++at)
            networkBoundaries.push_back(networkBoundaries.back() + std::stod(text.substr(at + 8)));
        const int64_t videoOrigin = split ? delivered.lower_bound({1, INT64_MIN})->first.second : 0;
        for (int id : {3, 0, 2, 1}) {
            auto response = request(base + "segment/" + std::to_string(id) + ".ts");
            assert(response.status == 200);
            const auto path = (directory / "network-output.ts").string(); writeFile(path, response.bytes);
            auto rows = samples(path, 1);
            Samples expected;
            for (const auto &[key, value] : delivered) {
                const int64_t shift = key.first ? videoOrigin : 0;
                const auto adjusted = std::pair{key.first, key.second - shift};
                if (adjusted.second / 90000.0 >= networkBoundaries[id] - .00002 &&
                    adjusted.second / 90000.0 < networkBoundaries[id + 1] - .00002)
                    expected.emplace(adjusted, CompressedSample{value.dts - shift, value.bytes});
            }
            if (rows != expected)
                std::cerr << "remux-work split=" << split << " id=" << id << " actual=" << rows.size()
                          << " expected=" << expected.size() << std::endl;
            assert(rows == expected);
            assert(request(base + "segment/" + std::to_string(id) + ".ts").bytes == response.bytes);
        }
        assert(upstream.foreignRequests == 0);
        cast_media_destroy(media);
    }
    // AAC transcoding must stay continuous across non-six-second boundaries.
    const auto videoURL = upstream.url("/video.mp4"), audioURL = upstream.url("/audio.wav");
    media = start({videoURL.c_str(), "X-Video: video-secret\r\n", audioURL.c_str(), "X-Audio: audio-secret\r\n",
                   false, false, false, 0}, url, duration);
    base = url.substr(0, url.rfind('/') + 1);
    auto before = audioSpan(request(base + "segment/0.ts"), (directory / "audio-output.ts").string());
    auto after = audioSpan(request(base + "segment/1.ts"), (directory / "audio-output.ts").string());
    assert(std::abs(after.first - before.second) < 1.0 / 48000);
    cast_media_destroy(media);

    // Reordered video and GOPs beyond the duration ceiling keep the fixed-grid
    // fallback, including random-access independently decodable output.
    for (bool reordered : {true, false}) {
        const auto fallback = (directory / "fallback.mp4").string();
        const char *args[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", source.c_str(),
            "-c:v", "libx264", "-profile:v", "main", "-level:v", "4.1",
            "-bf", reordered ? "2" : "0", "-g", reordered ? "70" : "140",
            "-keyint_min", reordered ? "70" : "140", "-sc_threshold", "0",
            "-c:a", "copy", "-movflags", "+faststart", fallback.c_str(), nullptr};
        assert(posix_spawnp(&child, args[0], nullptr, nullptr, const_cast<char *const *>(args), environ) == 0);
        assert(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0);
        media = start({fallback.c_str(), "", nullptr, nullptr, true, false, false, 0}, url, duration);
        manifest = request(url); text.assign(manifest.bytes.begin(), manifest.bytes.end());
        assert(text.find("#EXT-X-TARGETDURATION:6\n") != std::string::npos);
        base = url.substr(0, url.rfind('/') + 1);
        auto response = request(base + "segment/3.ts"); assert(response.status == 200);
        const auto path = (directory / "fallback-output.ts").string(); writeFile(path, response.bytes);
        const auto rows = samples(path, 1);
        for (const auto &row : rows) assert(row.first.second >= 18 * 90000 && row.first.second < 24 * 90000);
        cast_media_destroy(media);
    }
}
// Exercise the HTTP relay, not an isolated codec: delayed video, resampled AAC,
// independent segment decoding, cache eviction, and random-to-sequential seeks.
static void codecWork(const std::filesystem::path &directory) {
    const auto source = (directory / "codec.mp4").string(), audio = (directory / "codec.wav").string();
    const char *args[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
        "-f", "lavfi", "-i", "testsrc2=size=1280x720:rate=30:duration=61",
        "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=44100:duration=61",
        "-c:v", "libx264", "-preset", "ultrafast", "-bf", "2", "-g", "210",
        "-an", source.c_str(), "-vn", "-c:a", "pcm_s16le", audio.c_str(), nullptr};
    pid_t child = 0;
    assert(posix_spawnp(&child, args[0], nullptr, nullptr, const_cast<char *const *>(args), environ) == 0);
    int status = 0; assert(waitpid(child, &status, 0) == child);
    assert(WIFEXITED(status) && WEXITSTATUS(status) == 0);
    Upstream upstream(readFile(source), readFile(audio)); upstream.start();
    const auto videoURL = upstream.url("/source.ts"), audioURL = upstream.url("/audio.wav");
    std::string url; double duration = 0;
    const auto beforeDecode = decodeCounts();
    const auto began = std::chrono::steady_clock::now();
    auto *media = start({videoURL.c_str(), "X-Video: video-secret\r\n", audioURL.c_str(),
        "X-Audio: audio-secret\r\n", false, false, false, 0}, url, duration);
    const auto prepared = std::chrono::steady_clock::now();
    const auto base = url.substr(0, url.rfind('/') + 1), output = (directory / "codec-output.ts").string();
    std::map<int, std::pair<double, double>> spans;
    double sequentialMS = 0, randomMS = 0, sequentialCPU = 0;
    for (int id : {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 0, 1, 10}) {
        const auto before = std::chrono::steady_clock::now();
        const auto cpuBefore = std::clock();
        const auto response = request(base + "segment/" + std::to_string(id) + ".ts");
        const auto elapsed = std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - before).count();
        const double cpuMS = 1000.0 * (std::clock() - cpuBefore) / CLOCKS_PER_SEC;
        if (spans.size() < 10) { sequentialMS += elapsed; sequentialCPU += cpuMS; }
        else randomMS += elapsed;
        assert(response.status == 200); writeFile(output, response.bytes);
        const auto rows = samples(output, 1);
        int videos = 0;
        for (const auto &[key, value] : rows) {
            assert(key.second >= id * 6 * 90000 && key.second < std::min(61, (id + 1) * 6) * 90000);
            if (key.first) ++videos;
        }
        assert(videos == (id == 10 ? 30 : 180));
        auto span = audioSpan(response, output);
        if (spans.contains(id - 1)) assert(std::abs(span.first - spans.at(id - 1).second) < 1.0 / 48000);
        if (spans.contains(id + 1)) assert(std::abs(spans.at(id + 1).first - span.second) < 1.0 / 48000);
        spans[id] = span;
        assert(request(base + "segment/" + std::to_string(id) + ".ts").bytes == response.bytes);
    }
    assert(upstream.foreignRequests == 0);
    cast_media_destroy(media);
    std::cout << "codec-work prepare_ms=" << std::chrono::duration<double, std::milli>(prepared - began).count()
              << " sequential_60s_ms=" << sequentialMS << " sequential_cpu_ms=" << sequentialCPU
              << " seek_sequence_ms=" << randomMS;
    decodeUsage(beforeDecode); std::cout << std::endl;
    // A long video gap must not turn bounded encoder lookahead into HTTP 503,
    // and resuming video after audio-only segments still needs an independent IDR.
    const auto sparsePath = (directory / "sparse.mp4").string();
    const char *sparseArgs[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
        "-i", source.c_str(), "-t", "25", "-vf", "select='lt(t,3)+gte(t,18)'",
        "-fps_mode", "vfr", "-c:v", "libx264", "-preset", "ultrafast", "-bf", "2",
        "-an", sparsePath.c_str(), nullptr};
    assert(posix_spawnp(&child, sparseArgs[0], nullptr, nullptr, const_cast<char *const *>(sparseArgs), environ) == 0);
    assert(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0);
    Upstream sparse(readFile(sparsePath), readFile(audio)); sparse.start();
    const auto sparseVideo = sparse.url("/source.ts"), sparseAudio = sparse.url("/audio.wav");
    media = start({sparseVideo.c_str(), "X-Video: video-secret\r\n", sparseAudio.c_str(),
        "X-Audio: audio-secret\r\n", false, false, false, 0}, url, duration);
    const auto sparseBase = url.substr(0, url.rfind('/') + 1);
    double audioEnd = NAN;
    for (int id = 0; id < 5; ++id) {
        const auto response = request(sparseBase + "segment/" + std::to_string(id) + ".ts");
        const auto span = audioSpan(response, output);
        if (id) assert(std::abs(span.first - audioEnd) < 1.0 / 48000);
        audioEnd = span.second;
        if (id == 0 || id >= 3) {
            const auto rows = samples(output, 1);
            const auto videos = std::count_if(rows.begin(), rows.end(), [](const auto &row) { return row.first.first == 1; });
            assert(videos == (id == 0 ? 90 : id == 3 ? 180 : 30));
        }
    }
    cast_media_destroy(media);
}
// Twelve-second, 30 fps video-only fixtures exercise each codec without
// regenerating sources. Separate PCM audio also keeps the AAC lane covered.
static void decodeWork(const std::filesystem::path &directory, const std::string &fixturePath) {
#ifndef CAST_MEDIA_CHECK
    assert(false && "Compile --decode-work with CAST_MEDIA_CHECK to observe actual decoding");
#endif
    AVFormatContext *format = nullptr;
    ok(avformat_open_input(&format, fixturePath.c_str(), nullptr, nullptr)); ok(avformat_find_stream_info(format, nullptr));
    const int videoIndex = av_find_best_stream(format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0); assert(videoIndex >= 0);
    const auto *video = format->streams[videoIndex]->codecpar;
    const std::string codec = avcodec_get_name(video->codec_id);
    avformat_close_input(&format);
    Upstream upstream(readFile(fixturePath), wav()); upstream.start();
    const auto videoURL = upstream.url("/source.ts"), audioURL = upstream.url("/audio.wav");
    const auto output = (directory / "decode-output.ts").string();
    const auto beforeDecode = decodeCounts();
    double wallMS = 0, cpuMS = 0;
    for (bool random : {false, true}) {
        std::string url; double duration = 0;
        auto *media = start({videoURL.c_str(), "X-Video: video-secret\r\n", audioURL.c_str(),
            "X-Audio: audio-secret\r\n", false, false, false, 0}, url, duration);
        assert(std::abs(duration - 12) < .1);
        const auto base = url.substr(0, url.rfind('/') + 1);
        std::map<int, std::pair<double, double>> spans;
        for (int at = 0; at < 2; ++at) {
            const int id = random ? 1 - at : at;
            const auto began = std::chrono::steady_clock::now(); const auto cpu = std::clock();
            const auto response = request(base + "segment/" + std::to_string(id) + ".ts");
            wallMS += std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - began).count();
            cpuMS += 1000.0 * (std::clock() - cpu) / CLOCKS_PER_SEC;
            assert(response.status == 200); writeFile(output, response.bytes);
            const auto rows = samples(output, 1);
            int videos = 0;
            for (const auto &[key, value] : rows) {
                assert(key.second >= id * 6 * 90000 && key.second < (id + 1) * 6 * 90000);
                if (key.first) ++videos;
            }
            assert(videos == 180);
            spans[id] = audioSpan(response, output);
            assert(request(base + "segment/" + std::to_string(id) + ".ts").bytes == response.bytes);
        }
        assert(std::abs(spans.at(1).first - spans.at(0).second) < 1.0 / 48000);
        cast_media_destroy(media);
    }
    assert(upstream.foreignRequests == 0);
    std::cout << "decode-work codec=" << codec << " relay_24s_ms=" << wallMS << " relay_cpu_ms=" << cpuMS;
    decodeUsage(beforeDecode);
    std::cout << "\ncast-media: codec decode, independent IDRs, AAC continuity, sequential and random seeks passed\n";
}

// External fixtures keep upstream serving CPU/RSS out of the relay process.
// Cache seek means retained generated TS, not a packet-cache seek. All output
// decoding happens after the final snapshot, so validation cannot inflate any
// phase's CPU or process high-water RSS. Only endpoint responses are retained.
static void playbackBenchmark(std::string origin, const std::string &kind) {
    if (origin.ends_with('/')) origin.pop_back();
    const std::string prefix = "http://127.0.0.1:";
    assert(origin.starts_with(prefix));
    assert(origin.size() > prefix.size() && origin.find_first_not_of("0123456789", prefix.size()) == std::string::npos);
    const int port = std::stoi(origin.substr(prefix.size())); assert(port > 0 && port <= 65535);
    assert(kind == "split" || kind == "hls");
    const bool split = kind == "split";
    const std::string video = origin + (split ? "/media/bench-video.mp4" : "/media/bench.m3u8");
    const std::string audio = origin + "/media/bench-audio.m4a";
    const CastMediaSource source{video.c_str(), split ? "X-Terebi-Fixture: video\r\n" : "X-Terebi-Fixture: combined\r\n",
        split ? audio.c_str() : nullptr, split ? "X-Terebi-Fixture: audio\r\n" : nullptr, false, !split, false, 120};
    auto acknowledge = [] {
        std::string line; assert(bool(std::getline(std::cin, line)));
    };
    auto snapshot = [] {
        rusage usage{}; assert(getrusage(RUSAGE_SELF, &usage) == 0);
        return std::pair{std::chrono::steady_clock::now(), usage};
    };
    auto begin = [&](const char *phase) {
        std::cout << "{\"event\":\"begin\",\"phase\":\"" << phase << "\"}" << std::endl;
        acknowledge();
        return snapshot();
    };
    auto cpuMS = [](const rusage &usage) {
        return 1000.0 * (usage.ru_utime.tv_sec + usage.ru_stime.tv_sec)
            + (usage.ru_utime.tv_usec + usage.ru_stime.tv_usec) / 1000.0;
    };
    auto wallMS = [](auto from, auto to) {
        return std::chrono::duration<double, std::milli>(to - from).count();
    };
    auto finish = [&](const char *phase, const auto &before, const auto &after, double seconds, size_t bytes, const std::string &extra = "") {
#ifdef __APPLE__
        const double rssMiB = double(after.second.ru_maxrss) / (1024 * 1024);
#else
        const double rssMiB = double(after.second.ru_maxrss) / 1024;
#endif
        std::cout << "{\"phase\":\"" << phase << "\",\"wall_ms\":" << wallMS(before.first, after.first)
                  << ",\"cpu_ms\":" << cpuMS(after.second) - cpuMS(before.second)
                  << ",\"peak_rss_mib\":" << rssMiB << ",\"media_seconds\":" << seconds
                  << ",\"output_bytes\":" << bytes << extra << "}" << std::endl;
        acknowledge();
    };
    std::map<int, HTTP> outputs;
    std::string url; double duration = 0;
    const auto cold = begin("cold_start");
    auto *media = start(source, url, duration);
    const auto prepared = std::chrono::steady_clock::now();
    const auto manifest = request(url);
    const auto base = url.substr(0, url.rfind('/') + 1);
    const auto firstBegan = std::chrono::steady_clock::now();
    outputs.emplace(0, request(base + "segment/0.ts"));
    const auto coldEnd = snapshot();
    finish("cold_start", cold, coldEnd, 0, outputs.at(0).bytes.size(),
        ",\"prepare_ms\":" + std::to_string(wallMS(cold.first, prepared))
        + ",\"manifest_ms\":" + std::to_string(wallMS(prepared, firstBegan))
        + ",\"first_segment_ms\":" + std::to_string(wallMS(firstBegan, coldEnd.first)));
    assert(duration >= 120 && duration < 121 && manifest.status == 200);
    const std::string playlist(manifest.bytes.begin(), manifest.bytes.end());
    assert(playlist.find("#EXT-X-ENDLIST") != std::string::npos);
    std::vector<double> boundaries{0};
    for (size_t at = 0; (at = playlist.find("#EXTINF:", at)) != std::string::npos; ++at)
        boundaries.push_back(boundaries.back() + std::stod(playlist.substr(at + 8)));
    assert(boundaries.size() >= 18);
    for (int id = 0; id <= 17; ++id) assert(std::abs(boundaries[id] - id * 6) < .00002);

    // No intervening generation: segment zero is in the eight-entry TS cache.
    {
        const auto cached = begin("cached_seek");
        const auto repeated = request(base + "segment/0.ts");
        const auto cachedEnd = snapshot();
        finish("cached_seek", cached, cachedEnd, 0, repeated.bytes.size(), ",\"segment_id\":0");
        assert(repeated.status == 200 && repeated.bytes == outputs.at(0).bytes);
    }
    outputs.clear();

    // Sequential generation stays on the warm pipeline. The initialization and
    // first six seconds above are deliberately excluded; there is no pacing.
    size_t producedBytes = 0;
    const auto sustained = begin("sustained");
    for (int id = 1; id <= 10; ++id) {
        auto response = request(base + "segment/" + std::to_string(id) + ".ts");
        producedBytes += response.bytes.size();
        assert(response.status == 200);
        if (id == 10) outputs.emplace(id, std::move(response));
    }
    const auto sustainedEnd = snapshot();
    finish("sustained", sustained, sustainedEnd, boundaries[11] - boundaries[1], producedBytes,
        ",\"unpaced\":true,\"first_segment_excluded\":true,\"first_segment_id\":1,\"segment_count\":10");

    // Generating through 66s evicts segment zero from the eight-entry TS cache
    // and its packets from the thirty-second rewind history: a miss in both.
    const auto uncached = begin("uncached_seek");
    outputs.emplace(0, request(base + "segment/0.ts"));
    const auto uncachedEnd = snapshot();
    finish("uncached_seek", uncached, uncachedEnd, 0, outputs.at(0).bytes.size(), ",\"segment_id\":0");
    cast_media_destroy(media);

    char temp[] = "/tmp/terebi-cast-benchmark.XXXXXX"; assert(mkdtemp(temp));
    const auto output = (std::filesystem::path(temp) / "segment.ts").string();
    for (const auto &[id, response] : outputs) {
        assert(response.status == 200);
        writeFile(output, response.bytes);
        // samples requires an IDR entry, strict independent decode, both A/V,
        // monotonic DTS and one decoded frame for every compressed sample.
        const auto rows = samples(output, 1);
        int videos = 0, audios = 0;
        int64_t firstAudio = 0, firstVideo = 0;
        for (const auto &[key, value] : rows) {
            assert(key.second >= id * 6 * 90000 && key.second < (id + 1) * 6 * 90000);
            if (key.first) {
                if (!videos) {
                    firstVideo = key.second;
                    // The parent's HLS TS starts AAC 21.334ms before video;
                    // split MP4 starts both at zero. Allow TS rescale rounding.
                    const int64_t expected = id * 6 * 90000 + (split ? 0 : 1920);
                    assert(std::abs(firstVideo - expected) <= 2);
                }
                assert(key.second == firstVideo + videos * 3000);
                ++videos;
            } else {
                if (!audios) {
                    firstAudio = key.second;
                    assert(firstAudio < id * 6 * 90000 + 1922);
                }
                assert(key.second == firstAudio + audios * 1920);
                ++audios;
            }
        }
        assert(videos == 180 && audios >= 281 && audios <= 282);
    }
    std::filesystem::remove_all(temp);
    std::cerr << "cast playback benchmark: independently decodable IDRs, 30fps timestamps/frame counts and AAC verified\n";
}
int main(int argc, char **argv) {
    if (argc > 1 && std::string(argv[1]) == "--playback-benchmark") {
        if (argc != 4 || (std::string(argv[3]) != "split" && std::string(argv[3]) != "hls")) {
            std::cerr << "usage: " << argv[0] << " --playback-benchmark <base_url> <split|hls>\n";
            return 2;
        }
        playbackBenchmark(argv[2], argv[3]);
        return 0;
    }
    bool liveCheck = false, seekCheck = false, hlsCheck = false, remuxCheck = false;
    for (int argument = 1; argument < argc; ++argument) {
        if (std::string(argv[argument]) == "--live") liveCheck = true;
        if (std::string(argv[argument]) == "--seek-work") seekCheck = true;
        if (std::string(argv[argument]) == "--hls-work") hlsCheck = true;
        if (std::string(argv[argument]) == "--remux-work") remuxCheck = true;
        if (std::string(argv[argument]) == "--decode-work") {
            assert(argument + 1 < argc && "--decode-work requires a 12s/30fps fixture path");
            char temp[] = "/tmp/terebi-decode-check.XXXXXX"; assert(mkdtemp(temp));
            decodeWork(temp, argv[++argument]); std::filesystem::remove_all(temp);
            return 0;
        }
        if (std::string(argv[argument]) == "--codec-work") {
            char temp[] = "/tmp/terebi-codec-check.XXXXXX"; assert(mkdtemp(temp));
            codecWork(temp); std::filesystem::remove_all(temp);
            std::cout << "cast-media: sequential codec reuse, independent IDRs, AAC continuity and random seeks passed\n";
            return 0;
        }
    }
    char temp[] = "/tmp/terebi-cast-check.XXXXXX"; assert(mkdtemp(temp)); std::filesystem::path directory(temp);
    if (remuxCheck) {
        remuxWork(directory);
        std::filesystem::remove_all(directory);
        std::cout << "cast-media: keyframe-aligned indexed remux passed\n";
        return 0;
    }
    const auto video = (directory / "source.ts").string(), audio = (directory / "source.wav").string(), segment = (directory / "segment.ts").string();
    fixture(video); writeFile(audio, wav());
    const auto splitVideo = (directory / "video.mp4").string(), splitAudio = (directory / "audio.m4a").string();
    splitFixture(video, splitVideo, AVMEDIA_TYPE_VIDEO);
    splitFixture(video, splitAudio, AVMEDIA_TYPE_AUDIO);
    auto videoBytes = readFile(video), audioBytes = readFile(audio);
    Upstream upstream(videoBytes, audioBytes), foreign(videoBytes, audioBytes);
    upstream.resources["/video.mp4"] = readFile(splitVideo);
    upstream.resources["/audio.m4a"] = readFile(splitAudio);
    upstream.foreign = foreign.url(""); upstream.start(); foreign.start();
    seekWork(upstream, segment);
    if (seekCheck) {
        std::filesystem::remove_all(directory);
        std::cout << "cast-media: split VOD seek work passed\n";
        return 0;
    }
    const auto selectedHLS = (directory / "selected.ts").string(), unusedHLS = (directory / "unused.ts").string();
    fixture(selectedHLS, 180);
    fixture(unusedHLS, 180, 48, 32, 100000);
    Upstream delayed(videoBytes, audioBytes);
    delayed.resources.merge(hlsFixture(selectedHLS, unusedHLS, directory / "hls"));
    delayed.gated = true; delayed.start();
    hlsWork(delayed, segment);
    if (hlsCheck) {
        std::filesystem::remove_all(directory);
        std::cout << "cast-media: authenticated delayed multi-variant combined HLS passed\n";
        return 0;
    }
    std::string url; double duration = 0;
    CastMediaSource local{video.c_str(), "", nullptr, nullptr, true, false, false, 0};
    auto *media = start(local, url, duration); assert(duration >= 89 && duration <= 92);
    std::string base = url.substr(0, url.rfind('/') + 1);
    auto playlist = request(url); assert(playlist.status == 200); std::string text(playlist.bytes.begin(), playlist.bytes.end()); assert(text.find("#EXT-X-ENDLIST") != std::string::npos);
    playable(request(base + "segment/10.ts"), segment, true, 60);
    playable(request(base + "segment/0.ts"), segment, true, 0);
    assert(request(base + "../source.ts").status == 404);
    assert(request(base + "%2e%2e/source.ts").status == 404);
    assert(request(url.substr(0, url.find('/', 7)) + "/index.m3u8").status == 404);
    assert(request(base + "segment/999999.ts").status == 404);
    const char vtt[] = "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nHello\n"; char subtitle[512]{};
    assert(cast_media_subtitle(media, 1, vtt, sizeof(vtt) - 1, subtitle));
    auto subtitles = request(subtitle); assert(subtitles.status == 200 && subtitles.headers.find("Access-Control-Allow-Origin: *") != std::string::npos);
    auto ranged = request(subtitle, "GET", "Range: bytes=0-5\r\n"); assert(ranged.status == 206 && ranged.bytes.size() == 6);
    assert(request(subtitle, "HEAD").bytes.empty()); assert(request(subtitle, "OPTIONS").status == 204);
    assert(request(subtitle, "GET", "Range: bytes=99999-\r\n").status == 416);
    cast_media_cancel(media); cast_media_destroy(media);
    // Previously published capabilities expire, including the bound listening socket.
    auto authority = url.substr(7, url.find('/', 7) - 7); auto colon = authority.find(':'); sockaddr_in closed{};
    closed.sin_family = AF_INET; closed.sin_addr.s_addr = htonl(INADDR_LOOPBACK); closed.sin_port = htons(std::stoi(authority.substr(colon + 1)));
    int socket = ::socket(AF_INET, SOCK_STREAM, 0); assert(connect(socket, reinterpret_cast<sockaddr *>(&closed), sizeof(closed)) != 0); close(socket);
    std::string hls = upstream.url("/index.m3u8"); CastMediaSource miruro{hls.c_str(), "X-Video: video-secret\r\n", nullptr, nullptr, false, true, false, 90};
    media = start(miruro, url, duration); base = url.substr(0, url.rfind('/') + 1);
    playable(request(base + "segment/10.ts"), segment, true, 60); cast_media_destroy(media);
    std::string networkVideo = upstream.url("/source.ts"), networkAudio = upstream.url("/audio.wav");
    CastMediaSource youtube{networkVideo.c_str(), "X-Video: video-secret\r\n", networkAudio.c_str(), "X-Audio: audio-secret\r\n", false, false, false, 90};
    media = start(youtube, url, duration); base = url.substr(0, url.rfind('/') + 1);
    playable(request(base + "segment/10.ts"), segment, true, 60);
    auto vodFirstAudio = audioSpan(request(base + "segment/0.ts"), segment);
    auto vodSecondAudio = audioSpan(request(base + "segment/1.ts"), segment);
    assert(std::abs(vodSecondAudio.first - vodFirstAudio.second) < 1.0 / 48000);
    cast_media_destroy(media);
    assert(upstream.videoRequests > 0 && upstream.audioRequests > 0 && upstream.foreignRequests == 0);
    CastMediaSource audioOnly{audio.c_str(), "", nullptr, nullptr, true, false, false, 0};
    media = start(audioOnly, url, duration); base = url.substr(0, url.rfind('/') + 1); playable(request(base + "segment/10.ts"), segment, false, 60); cast_media_destroy(media);
    for (const auto &path : {"/redirect", "/cross.m3u8"}) {
        auto resource = upstream.url(path); CastMediaSource unsafe{resource.c_str(), "X-Video: video-secret\r\n", nullptr, nullptr, false, true, false, 90};
        char error[512]{}, result[512]{}; media = cast_media_create(&unsafe, error); assert(media);
        assert(!cast_media_start(media, "127.0.0.1", result, &duration, error)); assert(std::string(error).find("secret") == std::string::npos); cast_media_destroy(media);
    }
    assert(foreign.foreignRequests == 0 && foreign.videoRequests == 0);
    auto publicManifest = upstream.url("/public-cross.m3u8");
    CastMediaSource publicHLS{publicManifest.c_str(), "X-Video: video-secret\r\n", nullptr, nullptr, false, false, false, 90};
    media = start(publicHLS, url, duration); base = url.substr(0, url.rfind('/') + 1);
    playable(request(base + "segment/0.ts"), segment, true, 0); cast_media_destroy(media);
    assert(foreign.videoRequests > 0 && foreign.foreignRequests == 0);
    if (liveCheck) {
        youtube.live = true; media = start(youtube, url, duration); base = url.substr(0, url.rfind('/') + 1); assert(duration == 0);
        auto firstAudio = audioSpan(request(base + "segment/0.ts"), segment);
        auto secondAudio = audioSpan(request(base + "segment/1.ts"), segment);
        assert(std::abs(secondAudio.first - firstAudio.second) < 1.0 / 48000);
#ifdef CAST_MEDIA_CHECK
        assert(cast_media_check_advance(media, 54));
#else
        assert(false && "Compile the permanent live check with CAST_MEDIA_CHECK");
#endif
        auto response = request(url); assert(response.status == 200); std::string text(response.bytes.begin(), response.bytes.end());
        auto at = text.find("#EXT-X-MEDIA-SEQUENCE:"); assert(at != std::string::npos);
        int64_t sequence = std::stoll(text.substr(at + 22));
        size_t count = 0, cursor = 0;
        while ((cursor = text.find("#EXTINF:", cursor)) != std::string::npos) { ++count; cursor += 8; }
        assert(count == 8 && text.find("#EXT-X-ENDLIST") == std::string::npos);
        int64_t last = sequence + int64_t(count) - 1;
        assert(sequence > 0 && request(base + "segment/0.ts").status == 404);
        playable(request(base + "segment/" + std::to_string(last) + ".ts"), segment, true, 0); cast_media_destroy(media);
    }
    std::filesystem::remove_all(directory);
    std::cout << "cast-media: playable A/V, VOD random seek, separate audio, authenticated delayed multi-variant HLS retention, HLS origin/header policy, subtitles, ranges, capabilities, gated cancellation" << (liveCheck ? ", bounded live window" : " (use --live for rolling window)") << " passed\n";
}