#include "streamplayer_window.h"
#include "videoplayer_window.h"
#include "video_frame_image.h"
#include "sound_manager.h"
#include "../transport/stream_bridge.h"
#include <QVBoxLayout>
#include <QHBoxLayout>
#include <QLabel>
#include <QPushButton>
#include <QSlider>
#include <QCloseEvent>
#include <QKeyEvent>
#include <QTransform>
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <chrono>
#include <algorithm>
#include <cstring>
#include <cerrno>

extern "C" {
#include "../../../lib/debug_config.h"
#include "../../../src/stream/stream.h"
#include "../../../src/stream/stream_playback_clock.h"
#include "../../../src/stream/stream_playback_audio.h"
#ifdef HAVE_FFMPEG
#include <libavcodec/avcodec.h>
#include <libswscale/swscale.h>
#endif
}

namespace {
int64_t nowUs() {
    return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}
std::atomic<uint64_t> nextId{0};
QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); }
}

struct StreamPlayerWindow::Playback {
    struct Video { QImage image; int64_t pts; };
    struct AudioGap {
        uint64_t count = 0, silentFrames = 0;
        unsigned requested = 0, received = 0;
        int64_t reserve = 0, feedAge = -1, workerAge = -1, audioPts = -1;
        size_t videoQueue = 0;
        bool held = false;
    } audioGap;
    uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
    int64_t createdAt = nowUs(), firstInputAt = 0;
    std::atomic<bool> running{true}, audioAvailable{false};
    std::atomic<bool> complete{false}, failed{false};
    std::atomic<float> volume{1.0f};
    float outputVolume = 0.0f;
    uint64_t audioGeneration = 0;
    std::thread worker;
    std::mutex lock;
    std::mutex wakeLock;
    std::condition_variable wake;
    uint64_t wakeVersion = 0, dataSignals = 0;
    std::deque<Video> videos;
    size_t videoBytes = 0;
    stream_playback_audio* sound = nullptr;
    stream_playback_clock adaptation{};
    stream_playback_state timing{};
    stream_media_status media{};
    int64_t position = -1, clockAt = 0, ready = -1, audioDacAt = 0, deviceLead = 0, lastShownPts = -1;
    int64_t decoderLead = 0, framePeriod = 33333, lastInputPts = -1, lastDecodedPts = -1;
    uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX;
    bool started = false, presenting = false, videoReady = false, ended = false, videoEnded = false;
    bool activeAudio = false, holdAudio = false, resync = false;
    int64_t audioFedAt = 0, audioFedPts = -1, workerAt = 0, guiAt = 0, guiGapMax = 0;
    ~Playback() { stream_playback_audio_destroy(sound); }

    // Отдельный lock: уведомление из ядра не ждёт codec/PCM lock и не вызывает bridge.
    void notify(bool data = false) {
        { std::lock_guard<std::mutex> guard(wakeLock); wakeVersion++; if (data) dataSignals++; }
        wake.notify_one();
    }

    uint64_t played(int64_t now) const {
        if (!sound || !audioDacAt) return 0;
        long double frames = (long double)audioFrameStart + (long double)(now - audioDacAt) * 48000 / 1000000;
        uint64_t count = frames <= 0 ? 0 : (uint64_t)frames;
        return std::min(count, stream_playback_audio_written(sound));
    }
    int64_t clock(int64_t now) {
        if (!started) return 0;
        if (!presenting) return position;
        if (activeAudio && sound && audioAvailable)
            return std::max(position, stream_playback_audio_position(sound, played(now)));
        if (holdAudio) return position;
        int64_t proposed = position + (now - clockAt) * adaptation.tempo_milli / 1000;
        return std::max(position, std::min(ready, proposed));
    }
};

StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent, bool embedded)
    : QWidget(parent), m_playback(std::make_unique<Playback>()) {
    m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer requested id=%llu source=%llx epoch=%llx",
        (unsigned long long)m_playback->id, (unsigned long long)source, (unsigned long long)epoch);
    setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500);
    setObjectName(QStringLiteral("streamPlayer_%1").arg(source));
    auto* layout = new QVBoxLayout(this); auto* title = new QLabel(name, this); layout->addWidget(title);
    m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status);
    m_render = new VideoRenderWidget(this, true); layout->addWidget(m_render, 1);
    auto* controls = new QHBoxLayout;
    auto* volume = new QSlider(Qt::Horizontal, this); volume->setRange(0, 100); volume->setValue(100);
    volume->setAccessibleName(text("Stream volume")); controls->addWidget(new QLabel(text("Volume"), this)); controls->addWidget(volume);
    auto* fullscreen = new QPushButton(text("Fullscreen"), this); controls->addWidget(fullscreen);
    auto* close = new QPushButton(text("Close player"), this); controls->addWidget(close); layout->addLayout(controls);
    if (embedded) {
        fullscreen->hide(); close->hide(); layout->setContentsMargins(3, 3, 3, 3); layout->setSpacing(2);
        title->setSizePolicy(QSizePolicy::Ignored, QSizePolicy::Preferred); title->setToolTip(name);
        m_status->setSizePolicy(QSizePolicy::Ignored, QSizePolicy::Preferred); setMinimumSize(0, 0);
    }
    connect(volume, &QSlider::valueChanged, this, [this](int value) { m_playback->volume = value / 100.0f; });
    connect(fullscreen, &QPushButton::clicked, this, [this] { if (isFullScreen()) showNormal(); else showFullScreen(); });
    connect(close, &QPushButton::clicked, this, &QWidget::close);
    connect(this, &StreamPlayerWindow::statusChanged, m_status, &QLabel::setText, Qt::QueuedConnection);
    attachAudio();
    connect(&m_timer, &QTimer::timeout, this, &StreamPlayerWindow::tick); m_timer.start(15);
    m_playback->worker = std::thread(&StreamPlayerWindow::decode, this);
}

StreamPlayerWindow::~StreamPlayerWindow() { stop(); }

bool StreamPlayerWindow::finished() const { return m_playback->complete; }
bool StreamPlayerWindow::failed() const { return m_playback->failed; }
int StreamPlayerWindow::drainTimeoutMs() const {
    auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
    return p.complete ? 0 : static_cast<int>(std::clamp<int64_t>((p.ready - p.clock(nowUs())) / 1000 + 2000, 2000, 30000));
}

/* Hardware callback заполняет только ближайшую порцию; source time тишины остаётся неподвижным. */
void StreamPlayerWindow::attachAudio() {
    if (!m_playback->running || m_audioAttached) return;
    m_playback->audioAvailable = SoundManager::instance()->outputAvailable();
    m_audioAttached = SoundManager::instance()->attachStream(m_playback->id,
        [this](float* output, unsigned frames, int64_t firstTimeUs, uint64_t generation) {
        int16_t pcm[960 * 2] = {};
        auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
        p.notify();
        int64_t now = nowUs();
        bool restored = !p.audioAvailable.exchange(true);
        if ((p.audioGeneration && p.audioGeneration != generation) || (restored && p.started)) p.resync = true;
        p.audioGeneration = generation;
        p.deviceLead = std::max<int64_t>(0, firstTimeUs - now);
        if (!p.sound || !p.started || p.resync) return;
        p.audioFrameStart = stream_playback_audio_written(p.sound); p.audioDacAt = firstTimeUs;
        int count = !p.presenting || p.holdAudio ? 0 : stream_playback_audio_pull(p.sound, pcm, frames, p.ended ? INT64_MAX : p.ready);
        if (count < 0) { p.resync = true; count = 0; }
        if ((unsigned)count < frames) {
            stream_playback_audio_silence(p.sound, frames - (unsigned)count);
            if (p.presenting && !p.ended) {
                p.underruns++; p.audioGap.count++; p.audioGap.silentFrames += frames - (unsigned)count;
                p.audioGap.requested = frames; p.audioGap.received = (unsigned)count; p.audioGap.held = p.holdAudio;
                p.audioGap.reserve = p.ready - p.position; p.audioGap.videoQueue = p.videos.size();
                p.audioGap.feedAge = p.audioFedAt ? now - p.audioFedAt : -1; p.audioGap.audioPts = p.audioFedPts;
                p.audioGap.workerAge = p.workerAt ? now - p.workerAt : -1;
            }
        }
        float volume = p.volume.load();
        for (int i = 0; i < count; i++) {
            p.outputVolume += std::clamp(volume - p.outputVolume, -1.0f / 480, 1.0f / 480);
            output[2 * i] += pcm[2 * i] * (p.outputVolume / 32768.0f);
            output[2 * i + 1] += pcm[2 * i + 1] * (p.outputVolume / 32768.0f);
        }
        if (p.ended && p.endFrames == UINT64_MAX && stream_playback_audio_empty(p.sound))
            p.endFrames = stream_playback_audio_written(p.sound);
    });
    if (!m_audioAttached) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio attach failed id=%llu", (unsigned long long)m_playback->id);
        m_playback->audioAvailable = false;
    }
}

/* GUI держит только ближайшие готовые изображения; накопленный поток остаётся в RS-кольце. */
void StreamPlayerWindow::tick() {
    auto& p = *m_playback;
    if (!SoundManager::instance()->outputAvailable()) p.audioAvailable = false;
    QImage image; bool finished = false, first = false;
    int64_t shownPts = -1, readyPts = -1, mediaPosition = -1, deviceLead = 0;
    {
        std::lock_guard<std::mutex> guard(p.lock);
        int64_t now = nowUs();
        if (p.guiAt) p.guiGapMax = std::max(p.guiGapMax, now - p.guiAt);
        p.guiAt = now;
        int64_t clock = p.clock(now);
        while (p.started && !p.videos.empty() && p.videos.front().pts <= clock) {
            p.videoBytes -= static_cast<size_t>(p.videos.front().image.sizeInBytes());
            p.lastShownPts = p.videos.front().pts; image = std::move(p.videos.front().image); p.videos.pop_front();
        }
        if (!image.isNull()) {
            first = p.shown++ == 0;
            shownPts = p.lastShownPts; readyPts = p.ready; mediaPosition = clock; deviceLead = p.deviceLead;
        }
        bool audioDone = !p.activeAudio || !p.audioAvailable || (p.endFrames != UINT64_MAX && p.played(nowUs()) >= p.endFrames);
        finished = p.ended && p.videoEnded && p.videos.empty() && audioDone;
    }
    p.notify();
    if (!image.isNull()) {
        m_render->setFrame(image); m_status->setText(text("Live"));
        if (first) DEBUG_INFO(DEBUG_CATEGORY_STREAM,
            "desktop first frame id=%llu source=%llx epoch=%llx %dx%d RGB_stride=%lld stage=display_submit "
            "elapsed_us=%lld pts=%lld clock_us=%lld ready_us=%lld ready_ahead_us=%lld av_us=%lld device_lead_us=%lld",
            (unsigned long long)p.id, (unsigned long long)p.source, (unsigned long long)p.epoch,
            image.width(), image.height(), (long long)image.bytesPerLine(), (long long)(nowUs() - p.createdAt),
            (long long)shownPts, (long long)mediaPosition, (long long)readyPts, (long long)(readyPts - shownPts),
            (long long)(shownPts - mediaPosition), (long long)deviceLead);
        emit framePresented();
    }
    if (finished || p.failed) {
        if (finished) m_status->setText(text("Stream ended"));
        p.complete = true;
        SoundManager::instance()->detachStream(p.id); m_audioAttached = false; m_timer.stop();
    }
}

/* Callback устройства прекращается до освобождения DSP; worker снимает только свой reader. */
void StreamPlayerWindow::stop() {
    if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return;
    m_playback->notify();
    m_timer.stop(); gui_stream_stop(m_playback->id);
    SoundManager::instance()->detachStream(m_playback->id); m_audioAttached = false; m_playback->audioAvailable = false;
    if (m_playback->worker.joinable()) m_playback->worker.join();
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop player stopped id=%llu shown=%llu underruns=%llu",
               (unsigned long long)m_playback->id, (unsigned long long)m_playback->shown, (unsigned long long)m_playback->underruns);
}

void StreamPlayerWindow::closeEvent(QCloseEvent* event) { stop(); QWidget::closeEvent(event); }
void StreamPlayerWindow::keyPressEvent(QKeyEvent* event) {
    if (event->key() == Qt::Key_Escape) { if (isFullScreen()) showNormal(); else close(); }
    else QWidget::keyPressEvent(event);
}

/* Worker заполняет codec по готовности входа; GUI показывает только ближайшие декодированные кадры. */
void StreamPlayerWindow::decode() {
    auto& p = *m_playback;
#ifndef HAVE_FFMPEG
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop live decoder unavailable: FFmpeg disabled");
    p.failed = p.complete = true;
    emit statusChanged(text("Live video playback requires FFmpeg"));
#else
    AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc();
    SwsContext* converter = nullptr;
    bool waitKey = true, opened = false, finishing = false, eosSent = false;
    stream_media_format videoFormat{};
    std::deque<StreamRecord> records; // Только worker: кодированные записи двух треков, управление — барьер.
    uint32_t revision = 0;
    uint64_t decoded = 0, discarded = 0, reportedShown = 0;
    int64_t deadline = nowUs() + 10000000, report = nowUs(), decoderProgress = -1, lastInputAt = -1;
    uint64_t reportedGaps = 0;
    int64_t gapReportAt = 0;
    struct {
        uint64_t loops = 0, video = 0, audio = 0, empty = 0, batches = 0, batchMax = 0;
        int64_t codec = 0, codecMax = 0, convert = 0, convertMax = 0;
        int64_t audioFeed = 0, audioFeedMax = 0;
        int64_t pull = 0, pullMax = 0, wait = 0, waitMax = 0, loopMax = 0;
    } work;
    int64_t lastLoopAt = report;
    auto reset = [&](bool request) {
        records.clear();
        avcodec_free_context(&decoder); av_packet_unref(packet); waitKey = true; finishing = eosSent = false;
        decoderProgress = lastInputAt = -1;
        {
            std::lock_guard<std::mutex> guard(p.lock);
            stream_playback_audio_destroy(p.sound); p.sound = nullptr;
            p.videos.clear(); p.videoBytes = 0; p.started = p.presenting = p.videoReady = false; p.position = p.ready = -1;
            p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1;
            p.ended = p.videoEnded = false;
            p.audioDacAt = 0; p.audioFrameStart = 0; p.holdAudio = p.resync = false; p.endFrames = UINT64_MAX;
            p.audioFedAt = 0; p.audioFedPts = -1;
        }
        if (request && !gui_stream_resync(p.id)) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop playback resync refused");
        deadline = nowUs() + 10000000; emit statusChanged(text("Waiting for a key frame…"));
    };
    auto fail = [&](const QString& reason, bool retry = true) {
        p.failed = retry; p.complete = true;
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason));
        emit statusChanged(reason);
    };
    auto observe = [&] {
        stream_media_status status{};
        if (!gui_stream_status(p.id, status)) return;
        std::lock_guard<std::mutex> guard(p.lock);
        p.media = status;
        if (!status.initialized) return;
        if (!p.started) {
            p.activeAudio = status.audio_enabled && p.audioAvailable;
            if (p.activeAudio) p.sound = stream_playback_audio_create(status.start_us, 40);
            if (p.activeAudio && !p.sound) { p.running = false; return; }
            p.position = status.start_us; p.clockAt = nowUs(); p.started = true;
            stream_playback_clock_reset(&p.adaptation, p.id);
            DEBUG_INFO(DEBUG_CATEGORY_STREAM,
                "desktop playback starts id=%llu pts=%lld fixed_prefill=0 elapsed_us=%lld "
                "ready_us=%lld video_us=%lld audio_us=%lld initial_reserve_us=%lld audio=%d",
                (unsigned long long)p.id, (long long)p.position, (long long)(nowUs() - p.createdAt),
                (long long)status.ready_us, (long long)status.video_us, (long long)status.audio_us,
                (long long)((p.activeAudio ? status.ready_us : status.video_us) - p.position), p.activeAudio);
        }
        p.ready = p.activeAudio ? status.ready_us : status.video_us;
    };
    /* Забираем выход прежде следующего входа. При FIN тот же проход дренирует codec до EOF. */
    auto drainVideo = [&] {
        if (!decoder) return;
        for (unsigned attempt = 0; attempt < 8 && p.running; attempt++) {
            { std::lock_guard<std::mutex> guard(p.lock); if (p.videos.size() >= 2) return; }
            int64_t codecAt = nowUs();
            int result = avcodec_receive_frame(decoder, frame);
            int64_t codecUs = nowUs() - codecAt;
            work.codec += codecUs; work.codecMax = std::max(work.codecMax, codecUs);
            if (result == AVERROR(EAGAIN)) return;
            if (result == AVERROR_EOF) {
                std::lock_guard<std::mutex> guard(p.lock); p.videoEnded = true;
                DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop decoder drained id=%llu decoded=%llu", (unsigned long long)p.id,
                    (unsigned long long)decoded); return;
            }
            if (result < 0) {
                DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder output failed id=%llu codec=%u result=%d",
                    (unsigned long long)p.id, videoFormat.video_codec, result);
                fail(text("Stream video decoder failed")); p.running = false; return;
            }
            decoded++; decoderProgress = nowUs();
            if (frame->width <= 0 || frame->width > 4096 || frame->height <= 0 || frame->height > 4096) {
                DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder invalid size=%dx%d", frame->width, frame->height);
                fail(text("Invalid stream video dimensions")); p.running = false; return;
            }
            int64_t pts;
            { std::lock_guard<std::mutex> guard(p.lock);
              pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? p.lastInputPts : frame->best_effort_timestamp;
              p.decoderLead = std::max(p.decoderLead, p.lastInputPts - pts); p.lastDecodedPts = pts; p.videoReady = true;
              if (!p.presenting) { p.presenting = true; p.clockAt = nowUs();
                  DEBUG_INFO(DEBUG_CATEGORY_VIDEO,
                      "desktop first decoded frame id=%llu pts=%lld decoder_lead=%lld elapsed_us=%lld decode_us=%lld ready_ahead_us=%lld",
                      (unsigned long long)p.id, (long long)pts, (long long)p.decoderLead, (long long)(nowUs() - p.createdAt),
                      (long long)(p.firstInputAt ? nowUs() - p.firstInputAt : -1), (long long)(p.ready - pts)); }
              if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; } }
            int64_t convertAt = nowUs();
            converter = sws_getCachedContext(converter, frame->width, frame->height, (AVPixelFormat)frame->format,
                frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr);
            QImage image = video_frame_image(frame->width, frame->height);
            if (!converter || image.isNull()) { fail(text("Stream video conversion failed")); p.running = false; return; }
            uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr};
            int strides[] = {(int)image.bytesPerLine(), 0, 0, 0};
            if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) {
                fail(text("Stream video conversion failed")); p.running = false; return;
            }
            if (videoFormat.rotation) image = image.transformed(QTransform().rotate(videoFormat.rotation));
            int64_t convertUs = nowUs() - convertAt;
            work.convert += convertUs; work.convertMax = std::max(work.convertMax, convertUs);
            std::lock_guard<std::mutex> guard(p.lock);
            p.videoBytes += (size_t)image.sizeInBytes(); p.videos.push_back({std::move(image), pts});
        }
    };
    if (!frame || !packet) fail(text("Stream decoder allocation failed"));
    else if (!gui_stream_open(p.id, p.group, p.source, p.epoch,
        [](void* arg) { static_cast<Playback*>(arg)->notify(true); }, &p)) fail(text("Could not connect to the stream"));
    else while (p.running) {
        uint64_t wakeVersion;
        { std::lock_guard<std::mutex> guard(p.wakeLock); wakeVersion = p.wakeVersion; }
        int64_t loopAt = nowUs();
        work.loops++; work.loopMax = std::max(work.loopMax, loopAt - lastLoopAt); lastLoopAt = loopAt;
        if (loopAt - report >= 1000000) {
            double seconds = (loopAt - report) / 1000000.0;
            std::lock_guard<std::mutex> guard(p.lock);
            uint64_t dataSignals;
            { std::lock_guard<std::mutex> wakeGuard(p.wakeLock); dataSignals = p.dataSignals; p.dataSignals = 0; }
            DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
                "desktop viewer work id=%llu interval_us=%lld loops=%llu video=%llu audio=%llu empty=%llu "
                "codec_us=%lld/%lld convert_us=%lld/%lld audio_feed_us=%lld/%lld pull_us=%lld/%lld wait_us=%lld/%lld "
                "loop_max_us=%lld gui_gap_max_us=%lld gui_age_us=%lld decoded=%llu shown=%llu show_fps=%.1f "
                "clock_us=%lld ready_us=%lld videoQ=%zu hold=%d underruns=%llu batches=%llu batch_max=%llu "
                "pending_records=%zu data_signals=%llu presenting=%d audio_input=%d input_FIN=%d/%d/%d "
                "drain=%d/%d packet_bytes=%d timing=total/max",
                (unsigned long long)p.id, (long long)(loopAt - report), (unsigned long long)work.loops,
                (unsigned long long)work.video, (unsigned long long)work.audio, (unsigned long long)work.empty,
                (long long)work.codec, (long long)work.codecMax, (long long)work.convert, (long long)work.convertMax,
                (long long)work.audioFeed, (long long)work.audioFeedMax,
                (long long)work.pull, (long long)work.pullMax, (long long)work.wait, (long long)work.waitMax,
                (long long)work.loopMax, (long long)p.guiGapMax, (long long)(p.guiAt ? nowUs() - p.guiAt : -1),
                (unsigned long long)decoded, (unsigned long long)p.shown, (p.shown - reportedShown) / seconds,
                (long long)p.position, (long long)p.ready, p.videos.size(), p.holdAudio, (unsigned long long)p.underruns,
                (unsigned long long)work.batches, (unsigned long long)work.batchMax, records.size(), (unsigned long long)dataSignals,
                p.presenting, p.sound ? stream_playback_audio_needs_input(p.sound) : -1,
                p.media.finished, p.media.video_ended, p.media.audio_ended, finishing, eosSent, packet->size);
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
                "desktop viewer id=%llu decoded=%llu discarded=%llu shown=%llu showFps=%.1f videoQ=%zu "
                "clock=%lld reserve_us=%lld minimum_us=%lld tempo=%d lead_us=%lld underruns=%llu "
                "ready_us=%lld shown_pts=%lld display_lag_us=%lld decoder_lead_us=%lld hold=%d",
                (unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded,
                (unsigned long long)p.shown, (p.shown - reportedShown) / seconds, p.videos.size(), (long long)p.position,
                (long long)p.timing.reserve_us, (long long)p.timing.minimum_us, p.timing.tempo_milli,
                (long long)p.deviceLead, (unsigned long long)p.underruns, (long long)p.ready,
                (long long)p.lastShownPts, (long long)(p.lastShownPts >= 0 ? p.position - p.lastShownPts : -1),
                (long long)p.decoderLead, p.holdAudio);
            work = {}; report = loopAt; reportedShown = p.shown; p.guiGapMax = 0;
        }
        bool resync;
        Playback::AudioGap gap;
        { std::lock_guard<std::mutex> guard(p.lock); resync = p.resync; p.workerAt = nowUs(); gap = p.audioGap; }
        /* Callback сохраняет снимок; печать выполняется worker без блокировки аудиоустройства. */
        if (gap.count != reportedGaps && nowUs() - gapReportAt >= 1000000) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM,
                "desktop audio gap id=%llu count=%llu delta=%llu silence_frames=%llu cause=%s got=%u/%u "
                "reserve_us=%lld feed_age_us=%lld worker_age_us=%lld audio_pts=%lld videoQ=%zu",
                (unsigned long long)p.id, (unsigned long long)gap.count, (unsigned long long)(gap.count - reportedGaps),
                (unsigned long long)gap.silentFrames, gap.held ? "video_hold" : "pcm_short", gap.received, gap.requested,
                (long long)gap.reserve, (long long)gap.feedAge, (long long)gap.workerAge, (long long)gap.audioPts, gap.videoQueue);
            reportedGaps = gap.count; gapReportAt = nowUs();
        }
        if (resync) reset(true);
        observe();
        drainVideo();
        if (!p.running) break;
        bool space;
        { std::lock_guard<std::mutex> guard(p.lock); space = p.videos.size() < 2; }
        if (space && decoder && (packet->size || (finishing && !eosSent))) {
            int64_t codecAt = nowUs();
            int sent = avcodec_send_packet(decoder, packet->size ? packet : nullptr);
            int64_t codecUs = nowUs() - codecAt;
            work.codec += codecUs; work.codecMax = std::max(work.codecMax, codecUs);
            if (sent == 0) {
                if (packet->size) {
                    std::lock_guard<std::mutex> guard(p.lock);
                    if (p.lastInputPts >= 0) p.framePeriod = std::clamp<int64_t>(packet->pts - p.lastInputPts, 1, 1000000);
                    p.lastInputPts = packet->pts; lastInputAt = nowUs();
                } else eosSent = true;
                av_packet_unref(packet); drainVideo();
            } else if (sent != AVERROR(EAGAIN)) {
                DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder input failed id=%llu codec=%u bytes=%d result=%d",
                    (unsigned long long)p.id, videoFormat.video_codec, packet->size, sent);
                fail(text("Stream video decoder failed")); break;
            }
        }
        { std::lock_guard<std::mutex> guard(p.lock);
          if (finishing && p.videoEnded) break;
          if (p.videos.empty() && decoderProgress >= 0 && nowUs() - decoderProgress > 10000000
              && (packet->size || finishing || (lastInputAt >= 0 && nowUs() - lastInputAt < 1000000))) {
              DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder stalled id=%llu input_PTS=%lld output_PTS=%lld pending=%d FIN=%d",
                  (unsigned long long)p.id, (long long)p.lastInputPts, (long long)p.lastDecodedPts, packet->size, finishing);
              fail(text("Stream video decoder failed")); break;
          } }
        int64_t position, audioAhead; bool wantVideo; unsigned videoBudget;
        {
            std::lock_guard<std::mutex> guard(p.lock);
            int64_t now = nowUs(); position = p.started ? p.clock(now) : -1;
            wantVideo = !packet->size && p.videos.size() < 2;
            videoBudget = static_cast<unsigned>(std::max(0, 2 - (int)p.videos.size() - (packet->size ? 1 : 0)));
            audioAhead = !p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound) ? p.deviceLead + 80000 : -1;
            if (p.presenting && p.ready >= 0) {
                uint64_t played = p.played(now);
                p.holdAudio = !p.videoReady || (p.videos.empty() && p.lastDecodedPts < position - p.framePeriod)
                    || (!p.videos.empty() && p.videos.front().pts < position - 40000);
                int virtualEmpty = !p.activeAudio && p.position + (now - p.clockAt) * p.adaptation.tempo_milli / 1000 > p.ready;
                if (!p.activeAudio && p.holdAudio) position = p.position;
                int starved = p.holdAudio || virtualEmpty || (p.sound && stream_playback_audio_starved(p.sound, played));
                int64_t guardUs = std::max(p.decoderLead + p.framePeriod, p.activeAudio ? p.deviceLead + 10000 : p.framePeriod);
                int64_t pending = p.sound ? stream_playback_audio_pending_gain(p.sound, played) : 0;
                if (stream_playback_clock_update(&p.adaptation, now, std::max(p.position, position), p.ready,
                                                 guardUs, pending, !p.ended && starved, &p.timing) < 0) p.resync = true;
                else {
                    p.position = std::max(p.position, position); p.clockAt = now;
                    if (p.sound) stream_playback_audio_set_tempo(p.sound, p.timing.tempo_milli);
                }
            }
        }
        // Каждый трек расходуется по своей готовности. GAP/FIN/CLOSED не обгоняют предыдущие media.
        auto next = records.end();
        unsigned pendingVideo = 0, pendingAudio = 0; bool barrier = false;
        for (auto it = records.begin(); it != records.end(); ++it) {
            bool video = it->kind == StreamRecordKind::Video, audio = it->kind == StreamRecordKind::Audio;
            pendingVideo += video; pendingAudio += audio;
            if (!video && !audio) { barrier = true; if (it == records.begin()) next = it; break; }
            if (next == records.end() && ((video && wantVideo) || (audio && audioAhead >= 0))) next = it;
        }
        if (next == records.end()) {
            size_t previous = records.size();
            int64_t pullAt = nowUs();
            bool received = !finishing && !barrier && gui_stream_pull_batch(p.id, records, position,
                videoBudget > pendingVideo ? videoBudget - pendingVideo : 0, pendingAudio ? -1 : audioAhead);
            int64_t pullUs = nowUs() - pullAt;
            work.pull += pullUs; work.pullMax = std::max(work.pullMax, pullUs);
            if (received) {
                work.batches++; work.batchMax = std::max<uint64_t>(work.batchMax, records.size() - previous); continue;
            }
            work.empty++;
            if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
            if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
            int64_t waitAt = nowUs();
            { std::unique_lock<std::mutex> guard(p.wakeLock);
              p.wake.wait_for(guard, std::chrono::milliseconds(50), [&] { return !p.running || p.wakeVersion != wakeVersion; }); }
            int64_t waitUs = nowUs() - waitAt;
            work.wait += waitUs; work.waitMax = std::max(work.waitMax, waitUs); continue;
        }
        StreamRecord record = std::move(*next); records.erase(next);
        if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
        if (record.kind == StreamRecordKind::Gap) { reset(false); continue; }
        if (record.kind == StreamRecordKind::Closed) {
            const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again."
                : record.reason == STREAM_CLOSE_MEMBER ? "The source is no longer a channel member"
                : record.reason == STREAM_CLOSE_SERVICE ? "Chat service is stopped"
                : record.reason == STREAM_CLOSE_GROUP ? "The channel is closed" : "Could not continue the stream";
            fail(text(reason), record.reason == STREAM_CLOSE_FAILED); break;
        }
        if (record.kind == StreamRecordKind::End) {
            std::lock_guard<std::mutex> guard(p.lock); p.ended = true; p.holdAudio = false;
            if (p.sound) stream_playback_audio_finish(p.sound);
            finishing = true; p.videoEnded = !decoder; continue;
        }
        observe();
        if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }
        if (record.kind == StreamRecordKind::Audio) {
            work.audio++;
            int64_t audioAt = nowUs();
            int result = 0;
            { std::lock_guard<std::mutex> guard(p.lock);
              if (p.sound) result = stream_playback_audio_feed(p.sound, (int64_t)record.pts,
                  reinterpret_cast<const uint8_t*>(record.data.constData()), (size_t)record.data.size());
              if (p.sound && result == 0) { p.audioFedAt = nowUs(); p.audioFedPts = (int64_t)record.pts; } }
            int64_t audioUs = nowUs() - audioAt;
            work.audioFeed += audioUs; work.audioFeedMax = std::max(work.audioFeedMax, audioUs);
            if (result == -2) reset(true);
            else if (result < 0) { fail(text("Stream audio decoder failed")); break; }
            continue;
        }
        work.video++;
        if (revision != record.format.revision) {
            avcodec_free_context(&decoder); waitKey = true; revision = record.format.revision;
            decoderProgress = lastInputAt = -1;
            { std::lock_guard<std::mutex> guard(p.lock); p.videos.clear(); p.videoBytes = 0; p.videoReady = false;
              p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1; }
            deadline = nowUs() + 10000000;
        }
        videoFormat = record.format;
        if (waitKey && !(record.flags & STREAM_MEDIA_KEY)) continue;
        const uint8_t* data = reinterpret_cast<const uint8_t*>(record.data.constData());
        const uint8_t* configuration = nullptr; size_t size = (size_t)record.data.size(), configurationSize = 0;
        if (record.flags & STREAM_MEDIA_KEY) {
            if (stream_media_key_parse(data, size, &configuration, &configurationSize, &data, &size) != STREAM_OK) {
                fail(text("Stream video decoder failed")); break;
            }
        }
        if (!decoder) {
            int64_t openingAt = nowUs();
            AVCodecID codecId = record.format.video_codec == STREAM_MEDIA_H264 ? AV_CODEC_ID_H264
                : record.format.video_codec == STREAM_MEDIA_H265 ? AV_CODEC_ID_HEVC
                : record.format.video_codec == STREAM_MEDIA_VP9 ? AV_CODEC_ID_VP9
                : record.format.video_codec == STREAM_MEDIA_AV1 ? AV_CODEC_ID_AV1 : AV_CODEC_ID_NONE;
            if (record.format.video_codec == STREAM_MEDIA_H266) {
                const AVCodecDescriptor* descriptor = avcodec_descriptor_get_by_name("vvc");
                if (descriptor) codecId = descriptor->id;
            }
            const AVCodec* codec = avcodec_find_decoder(codecId);
            decoder = codec ? avcodec_alloc_context3(codec) : nullptr;
            if (decoder) {
                decoder->thread_count = 1; decoder->flags |= AV_CODEC_FLAG_LOW_DELAY;
                if (configurationSize) {
                    decoder->extradata = static_cast<uint8_t*>(av_mallocz(configurationSize + AV_INPUT_BUFFER_PADDING_SIZE));
                    if (!decoder->extradata) { fail(text("Stream decoder allocation failed")); break; }
                    std::memcpy(decoder->extradata, configuration, configurationSize); decoder->extradata_size = (int)configurationSize;
                }
            }
            int openResult = !codec ? AVERROR_DECODER_NOT_FOUND : !decoder ? AVERROR(ENOMEM) : avcodec_open2(decoder, codec, nullptr);
            if (openResult < 0) {
                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop decoder open failed codec=%u name=%s config=%zu result=%d",
                            record.format.video_codec, codec ? codec->name : "unavailable", configurationSize, openResult);
                fail(text("Stream video decoder failed")); break;
            }
            DEBUG_INFO(DEBUG_CATEGORY_STREAM,
                "desktop decoder id=%llu codec=%u name=%s config=%zu elapsed_us=%lld open_us=%lld key_pts=%llu",
                (unsigned long long)p.id, record.format.video_codec, codec->name, configurationSize,
                (long long)(nowUs() - p.createdAt), (long long)(nowUs() - openingAt), (unsigned long long)record.pts);
            decoder->pkt_timebase = AVRational{1, 1000000}; waitKey = false;
            p.firstInputAt = nowUs();
        }
        av_packet_unref(packet);
        if (av_new_packet(packet, (int)size) < 0) { fail(text("Stream decoder allocation failed")); break; }
        std::memcpy(packet->data, data, size);
        packet->pts = packet->dts = (int64_t)record.pts;
        if (lastInputAt < 0 || nowUs() - lastInputAt > 1000000) decoderProgress = nowUs();
    }
    avcodec_free_context(&decoder); av_frame_free(&frame); av_packet_free(&packet);
    if (converter) sws_freeContext(converter);
#endif
    gui_stream_stop(p.id);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop decoder released id=%llu", (unsigned long long)p.id);
}