#include "consumption_diagnostic.h"
#include "source_ownership.h"
// sbstreamer.cpp -- HTTP request broker that serves the FLAC stream to Sonos
//
// Copyright (c) 2026 Jaap van Vliet
//
// Original implementation for the YeneY project.
// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
// Licensed under the PolyForm Noncommercial License 1.0.0. See LICENSE.
//
// THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.

#include "upnp/stale_stream.h"
#include "sbstreamer.h"
#include "timing_probe_runtime.h"
#include "stream_session.h"
#include "upnp/timing.h"
#include "stream_close_log.h"

#include "sbencoder.h"
#include "stream_metrics.h"
#include "stream_debug.h"
#include "audio_mode.h"
#include <map>
#include "sonos-position.h"

#include <atomic>
#include <cstring>
#include <algorithm>
#include <cerrno>
#include <chrono>
#include <memory>
#include <mutex>
#include <unistd.h>

#define SBSTREAMER_ICON "/pulseaudio.png"
#define SBSTREAMER_CONTENT "audio/flac"
#define SBSTREAMER_DESC "Audio stream from %s"
#define SBSTREAMER_TIMEOUT 10000
// Four seconds waiting + bounded socket writes/EOF stays below ~five seconds.
#define SBSTREAMER_HTTP_IDLE_TIMEOUT 4000
// Allow the device play -> LMS CLI -> strm u round trip a full five seconds.
#define SBSTREAMER_RESUME_TIMEOUT 5000
#define SBSTREAMER_STANDBY_TIMEOUT 30000
#define SBSTREAMER_CHUNK 16384

using namespace bridge;

// The LMS generation outlives every HTTP connection. Each ACTIVE request owns
// a fresh FLAC encoder in that generation; STANDBY requests have no encoder.
static std::shared_ptr<SBEncoder> g_enc;
static std::mutex g_enc_mutex;
static std::map<unsigned, unsigned> streamRates;
// Publish the next format before its ID, without holding g_enc_mutex while
// new_squeezebox_stream_id takes the transport/resume locks.
extern "C" void set_squeezebox_audio_rate(unsigned stream, unsigned rate) {
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    streamRates[stream] = rate;
    while (streamRates.size() > 2) streamRates.erase(streamRates.begin());
}
// Ownership is independent of socket state. All request fields and slots are
// protected by g_enc_mutex; only ACTIVE requests have an encoder.
struct StreamRequest {
    std::chrono::steady_clock::time_point arrived = std::chrono::steady_clock::now();
    unsigned long long id;
    unsigned stream;
    std::string restartTag;
    std::shared_ptr<SBEncoder> encoder;
    std::atomic<bool> replaced{false};
    bool opened = false;
    bool pauseEnded = false;
    std::chrono::steady_clock::time_point expectedCloseAt{};
    std::atomic<bool> heldResume{false};
    std::atomic<bool> resumeAcknowledged{false};
    std::atomic<bool> serving{false};
};
static bridge::StreamCloseLog closeLog;
static std::atomic<bool> deferCloseLog{false};
extern "C" void configure_squeezebox_close_logging(bool own) { deferCloseLog = own; }
static unsigned long long nextRequestId = 0;
static unsigned ownershipStream = 0;
static std::shared_ptr<StreamRequest> activeRequest;
static std::shared_ptr<SBEncoder> canonicalEncoder;
static std::string canonicalRestartTag;
static std::vector<std::shared_ptr<StreamRequest>> standbyRequests;

// Caller holds g_enc_mutex, including when promoting a standby. Publish the
// fresh encoder and its owner together so PCM always goes to the ACTIVE request.
static void activateRequest(const std::shared_ptr<StreamRequest>& request, bool promoted)
{
    if (g_enc) g_enc->retireProducer();
    request->encoder = std::make_shared<SBEncoder>(request->stream);
    const auto rate = streamRates.find(request->stream);
    request->opened = request->encoder->open(audioMode() == AudioMode::Legacy ? 16 : 24,
        rate == streamRates.end() ? 44100 : rate->second);
    sonos_position_connection(request->stream, request->id);
    activeRequest = request;
    g_enc = request->encoder;
    streamMetrics().encoder([weak = std::weak_ptr<SBEncoder>(g_enc), connection = request->id] {
        auto enc = weak.lock();
        if (!enc) return StreamMeasurement{};
        const auto rate = enc->sampleRate();
        double frame = 0, scale = 1;
        bool locked = streamPacingSonos() && sonos_stream_clock(enc->streamId(),rate,&frame,&scale);
        return StreamMeasurement{double(enc->frameOrigin.load())*1000/rate + enc->encodedAudioMs(),
            double(estimate_sonos_audible_frames(rate))*1000/rate, enc->hasAudio() && !enc->responseEnded(), connection, locked};
    });
    auto weak = std::weak_ptr<SBEncoder>(g_enc);
    g_enc->pacingClock([weak] {
        auto enc = weak.lock();
        if (!enc) return StreamClock{};
        double frame = 0, scale = 1;
        bool locked = sonos_stream_clock(enc->streamId(), enc->sampleRate(), &frame, &scale);
        return StreamClock{(frame-double(enc->frameOrigin.load()))*1000/enc->sampleRate(), scale, locked};
    });
    canonicalEncoder = g_enc;
    canonicalRestartTag = request->restartTag;
    if (promoted && deferCloseLog) closeLog.observe(request->stream);
    if (promoted) request->expectedCloseAt = std::chrono::steady_clock::now();
    if (promoted)
        printf("stream %u: GET #%llu promoted\n", request->stream, request->id);
    printf("stream %u: GET #%llu ACTIVE\n", request->stream, request->id);
}
static unsigned endedByPause = 0;
extern void ResumeSqueezeBox(unsigned current);
extern void ResumeSqueezeBoxGetPair(unsigned, unsigned long long, unsigned long long,
                                  std::chrono::steady_clock::duration);
extern std::string SqueezeBoxURL(unsigned current);
extern "C" unsigned get_squeezebox_stream_id(void);
extern "C" unsigned get_lms_stream_serial(void);
extern "C" int yeney_is_paused(void);
extern "C" int sonos_output_running(void);

// Signal only the HTTP lifetime. Its worker sends EOF and retains the encoder
// until the next same-ID GET replaces it. No generation or FLAC close here.
static void endSqueezeboxResponse(bool flush)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    endedByPause = get_squeezebox_stream_id();
    if (deferCloseLog) closeLog.observe(endedByPause);
    if (activeRequest) activeRequest->expectedCloseAt = std::chrono::steady_clock::now();
    const bool preserve = flush && activeRequest && activeRequest->heldResume
        && !activeRequest->serving;
    if (preserve) activeRequest->resumeAcknowledged = false;
    if (!preserve && g_enc && g_enc->streamId() == endedByPause) g_enc->endResponse();
    // A pause ends existing requests, not a handoff to a standby. Standbys
    // have sent no headers and disconnect silently; later GETs may wait for PCM.
    for (auto& request : standbyRequests)
        request->pauseEnded = true;
    standbyRequests.clear();
}

extern "C" void note_squeezebox_device_close(void)
{
    if (deferCloseLog) closeLog.observe(get_squeezebox_stream_id());
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    if (activeRequest) activeRequest->expectedCloseAt = std::chrono::steady_clock::now();
}

extern "C" {
void end_squeezebox_response(void) { endSqueezeboxResponse(false); }
void flush_squeezebox_response(void) { endSqueezeboxResponse(true); }

void hold_squeezebox_resume(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    if (activeRequest && activeRequest->stream == stream && !activeRequest->encoder->hasAudio()
        && !activeRequest->encoder->responseEnded() && !activeRequest->encoder->cancelled())
        activeRequest->heldResume = true;
}

int squeezebox_response_streaming(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    return activeRequest && activeRequest->stream == stream && activeRequest->serving
        && g_enc && g_enc->hasAudio() && !g_enc->cancelled() && !g_enc->responseEnded();
}

int squeezebox_request_open(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    if (activeRequest && activeRequest->stream == stream) return 1;
    for (const auto& request : standbyRequests) if (request->stream == stream) return 1;
    return 0;
}

int squeezebox_response_open(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    return activeRequest && activeRequest->stream == stream && g_enc && g_enc->streamId() == stream && !g_enc->cancelled()
        && !g_enc->responseEnded();
}

int squeezebox_response_ended(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    return stream != 0 && endedByPause == stream;
}

void acknowledge_squeezebox_resume(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    if (endedByPause == stream) endedByPause = 0;
    if (activeRequest && activeRequest->stream == stream)
        activeRequest->resumeAcknowledged = true;
}

void invalidate_squeezebox_held_get(unsigned stream)
{
    std::lock_guard<std::mutex> lock(g_enc_mutex);
    if (g_enc && g_enc->streamId() == stream && !g_enc->hasAudio() && !g_enc->responseEnded()) {
        printf("stream %u: invalidating held GET for same-URL resume\n", stream);
        g_enc->cancel();
    }
}

int encode_squeezebox_audio_cancellable(const char* data, int len, uint64_t firstFrame, int (*cancel)(void*), void* context)
{
    auto cancelled = [=] { return cancel && cancel(context); };
    unsigned stream = get_squeezebox_stream_id();
    unsigned serial = get_lms_stream_serial();
    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    while (!cancelled() && sonos_output_running() && stream == get_squeezebox_stream_id() && serial == get_lms_stream_serial()) {
        std::shared_ptr<SBEncoder> enc;
        uint64_t requestId = 0;
        {
            std::lock_guard<std::mutex> lock(g_enc_mutex);
            enc = g_enc;
            if (activeRequest && enc == activeRequest->encoder) requestId = activeRequest->id;
        }
        if (enc && enc->streamId() == stream && !enc->cancelled() && !enc->responseEnded() && !enc->producerRetired()) {
            if (!enc->hasAcceptedPcm()) enc->frameOrigin = firstFrame;
            int written = enc->write(data, len, SBSTREAMER_TIMEOUT, [=] {
                enc->frameOrigin = firstFrame;
                sonos_position_pcm(stream, requestId, firstFrame);
            }, [=] { return cancelled() || !sonos_output_running(); });
            if (written == len) {
                if (timing_probe::enabled()) timing_probe::diagnostics().handed(stream, firstFrame, enc->sampleRate());
                return 1;
            }
            if (cancelled() || !sonos_output_running()) return 0;
            // Active-request termination or resume may replace the encoder
            // while write waits. Retry the SAME PCM block on the new encoder.
            if (!enc->cancelled() && !enc->responseEnded() && !enc->producerRetired()) {
                printf("encode_squeezebox_audio: write() failed %d != %d\n", written, len);
                return 0;
            }
        }
        if (yeney_is_paused())
            deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
        if (std::chrono::steady_clock::now() >= deadline) {
            printf("encode_squeezebox_audio: timeout waiting for stream request\n");
            return 0;
        }
        usleep(1000);
    }
    return 0;
}
void encode_squeezebox_audio(const char* data, int len, uint64_t firstFrame) {
    encode_squeezebox_audio_cancellable(data, len, firstFrame, nullptr, nullptr);
}

} // extern "C"

void SBStreamer::registerWith(upnp::StreamServer& server)
{
    server.registerStream(SBSTREAMER_CNAME, SBSTREAMER_URI, SBSTREAMER_DESC,
        SBSTREAMER_CONTENT, SBSTREAMER_ICON,
        [this](upnp::StreamRequest& request) { return HandleRequest(&request); });
}

bool SBStreamer::HandleRequest(upnp::StreamRequest* handle)
{
    if (IsAborted() || handle->aborted())
        return false;

    const std::string& requestUri = handle->path();
    if (requestUri.compare(0, strlen(SBSTREAMER_URI), SBSTREAMER_URI) != 0)
        return false;

    const auto method = handle->method();
    if (method != upnp::StreamRequest::Method::Get && method != upnp::StreamRequest::Method::Head) return false;
    unsigned long long requestId = 0;
    if (method == upnp::StreamRequest::Method::Get) {
        { std::lock_guard<std::mutex> lock(g_enc_mutex); requestId = ++nextRequestId; }
        printf("stream %d: GET #%llu headers: %s\n", atoi(handle->parameter("stream").c_str()), requestId,
            upnp::streamHeaderLog(handle->headers()).c_str());
    }
    const std::string session = handle->parameter("session");
    if (session != streamSessionToken()) {
        printf("stale request: session %s != %s\n", session.c_str(), streamSessionToken().c_str());
        upnp::rejectStaleStream(*handle);
        return true;
    }

    switch (method) {
    case upnp::StreamRequest::Method::Get: {
        int stream = atoi(handle->parameter("stream").c_str());
        streamSqueezeBox(handle, stream, requestId);
        return true;
    }
    case upnp::StreamRequest::Method::Head: {
        handle->reply(200, SBSTREAMER_CONTENT);
        return true;
    }
    default:
        return false;  // unhandled method
    }
}

void SBStreamer::streamSqueezeBox(upnp::StreamRequest* handle, int stream, unsigned long long requestId)
{
    StreamDebugRequest diagnostics(*handle);
    handle = &diagnostics;
    printf("Sonos requested stream %d\n", stream);
    // Bound a stalled peer as well as a stalled PCM producer. Sending uses
    // the socket directly, so a receive timeout alone is insufficient.
    handle->sendTimeout(500);
    auto request = std::make_shared<StreamRequest>();
    {
        std::lock_guard<std::mutex> lock(g_enc_mutex);
        request->id = requestId;
        request->stream = stream;
        request->restartTag = handle->parameter("restart");
    }
    auto peerClosed = [handle, request] { return request->replaced.load() || handle->peerClosed(); };
    unsigned current = get_squeezebox_stream_id();
    if (stream <= 0 || (unsigned)stream > current) {
        handle->reply(400);
        return;
    }
    auto redirect = [&] {
        std::string url = SqueezeBoxURL(get_squeezebox_stream_id());
        if (!handle->header("Range").empty()) url += "&restart=" + std::to_string(requestId);
        std::string response = "HTTP/1.1 302 Found\r\nLocation: " + url
            + "\r\nContent-Length: 0\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n";
        printf("stream %d: HTTP 302 -> %s\n", stream, url.c_str());
        handle->send(response.c_str(), response.size());
    };
    if ((unsigned)stream < current) {
        redirect();
        return;
    }

    if (speakerRelinquished.load() && yeney_is_paused()) {
        // Sonos can restore the previous queue after AirPlay. Do not request
        // LMS resume and do not turn a paused queue into a 503 retry loop.
        const std::string paused = "HTTP/1.1 200 OK\r\nContent-Type: audio/flac\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
        handle->send(paused.c_str(), paused.size());
        return;
    }

    const auto range = handle->header("Range");
    bool forceRestart;
    {
        std::lock_guard<std::mutex> lock(g_enc_mutex);
        forceRestart = !request->restartTag.empty() && request->restartTag != canonicalRestartTag;
        if (forceRestart && activeRequest) { activeRequest->replaced = true; activeRequest.reset(); }
    }
    if (!range.empty() && !forceRestart) {
        uint64_t offset = 0;
        std::shared_ptr<SBEncoder> canonical;
        {
            std::lock_guard<std::mutex> lock(g_enc_mutex);
            canonical = canonicalEncoder;
        }
        auto restart = [&](const char* reason) {
            static std::mutex failureMutex;
            static auto last = std::chrono::steady_clock::time_point{};
            {
                std::lock_guard<std::mutex> lock(failureMutex);
                auto now = std::chrono::steady_clock::now();
                if (now - last >= std::chrono::seconds(5)) {
                    printf("stream %d: Range unavailable (%s) → restart\n", stream, reason); last = now;
                }
            }
            const auto url = SqueezeBoxURL(get_squeezebox_stream_id()) + "&restart=" + std::to_string(requestId);
            const auto response = "HTTP/1.1 302 Found\r\nLocation: " + url
                + "\r\nContent-Length: 0\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n";
            handle->send(response.data(), response.size());
        };
        if (!openByteRange(range, offset) || !canonical || canonical->streamId() != (unsigned)stream
            || canonical->cancelled() || canonical->responseEnded() || canonical->producerRetired()) {
            restart("invalid range or inactive history"); return;
        }
        char bytes[SBSTREAMER_CHUNK];
        auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500);
        int count;
        while ((count = canonical->history.read(offset, bytes, sizeof(bytes))) == 0
               && !peerClosed() && !handle->aborted() && !IsAborted()
               && !canonical->historyFinished() && !canonical->responseEnded()
               && (unsigned)stream == get_squeezebox_stream_id()
               && std::chrono::steady_clock::now() < deadline) usleep(1000);
        if (peerClosed() || handle->aborted() || IsAborted()) return;
        if (count <= 0 || (unsigned)stream != get_squeezebox_stream_id()) {
            restart(count < 0 ? "evicted offset" : "offset not produced within 500 ms"); return;
        }
        bool superseded;
        {
            std::lock_guard<std::mutex> lock(g_enc_mutex);
            superseded = canonical != canonicalEncoder || canonical->responseEnded() || canonical->cancelled()
                || (unsigned)stream != get_squeezebox_stream_id();
            if (!superseded) {
                if (activeRequest) activeRequest->replaced = true;
                request->encoder = canonical; request->opened = true; request->serving = true;
                activeRequest = request; g_enc = canonical; canonical->useHistoryReader();
            }
        }
        if (superseded) { restart("superseded history"); return; }
        diagnostics.encoder([canonical] { return canonical->encodedAudioMs(); }, true);
        const auto bounds = canonical->history.bounds();
        printf("stream %d: Range resume from %llu (history %llu–%llu) → continued\n", stream,
            (unsigned long long)offset, (unsigned long long)bounds.first, (unsigned long long)bounds.second);
        // Finite partial response, unknown total entity length. Never advertise
        // an open last-byte position or append bytes beyond the stated range.
        const uint64_t rangeEnd = bounds.second;
        const std::string headers = "HTTP/1.1 206 Partial Content\r\nContent-Type: audio/flac\r\n"
            "Accept-Ranges: bytes\r\nContent-Range: bytes " + std::to_string(offset) + "-"
            + std::to_string(rangeEnd - 1) + "/*\r\nContent-Length: " + std::to_string(rangeEnd - offset)
            + "\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n";
        ConsumptionDiagnostic consumption(*handle, playerName().trueName, "stream");
        bool sent = handle->send(headers.data(), headers.size());
        while (sent && offset < rangeEnd && !peerClosed() && !handle->aborted() && !IsAborted()
               && !canonical->cancelled() && !canonical->responseEnded() && !canonical->producerRetired()
               && (unsigned)stream == get_squeezebox_stream_id()) {
            count = canonical->history.read(offset, bytes, std::min<uint64_t>(sizeof(bytes), rangeEnd-offset));
            if (count <= 0) break; // slow receiver fell out of the bounded window
            sent = handle->send(bytes, count); offset += count;
        }
        {
            std::lock_guard<std::mutex> lock(g_enc_mutex);
            if (activeRequest == request) {
                activeRequest.reset();
                if (!yeney_is_paused() && (unsigned)stream == get_squeezebox_stream_id() && !standbyRequests.empty()) {
                    auto newest = std::max_element(standbyRequests.begin(), standbyRequests.end(),
                        [](const auto& a, const auto& b) { return a->id < b->id; });
                    auto promoted = *newest; standbyRequests.erase(newest); activateRequest(promoted, true);
                }
            }
        }
        handle->disconnect(); return;
    }
    // Recovery GET+Range pairs on firmware 86.10 send Connection: close.
    // Give that plain probe 100 ms to close before committing a new encoder.
    bool recoveryProbe = false;
    {
        std::lock_guard<std::mutex> lock(g_enc_mutex);
        recoveryProbe = canonicalEncoder && canonicalEncoder->hasAudio()
            && !canonicalEncoder->responseEnded() && !canonicalEncoder->cancelled()
            && canonicalEncoder->streamId() == (unsigned)stream && !yeney_is_paused()
            && handle->header("Connection") == "close";
    }
    if (recoveryProbe) {
        auto until = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
        while (!peerClosed() && !handle->aborted() && !IsAborted()
               && std::chrono::steady_clock::now() < until) usleep(1000);
        if (peerClosed() || handle->aborted() || IsAborted()) return;
    }

    unsigned long long pairActive = 0;
    std::chrono::steady_clock::duration pairSeparation{};
    {
        std::lock_guard<std::mutex> lock(g_enc_mutex);
        // A track change may have won the race since request validation.
        if ((unsigned)stream == get_squeezebox_stream_id()) {
            if (ownershipStream != (unsigned)stream) {
                if (g_enc) g_enc->cancel();
                activeRequest.reset();
                standbyRequests.clear(); // old workers redirect on their next poll
                ownershipStream = stream;
            }
            // Preserve pause-ended encoders until a later GET replaces them.
            if (activeRequest && activeRequest->encoder->responseEnded())
                activeRequest.reset();
            if (!activeRequest) {
                activateRequest(request, false);
            } else {
                standbyRequests.push_back(request);
                pairSeparation = request->arrived - activeRequest->arrived;
                if (pairSeparation >= std::chrono::steady_clock::duration::zero()
                    && pairSeparation <= upnp::timing::kGetPairWindow)
                    pairActive = activeRequest->id;
                printf("stream %d: GET #%llu STANDBY\n", stream, request->id);
            }
        }
    }

    if (pairActive) ResumeSqueezeBoxGetPair(stream, pairActive, request->id, pairSeparation);

    std::shared_ptr<SBEncoder> enc;
    bool opened = false;
    auto standbyDeadline = std::chrono::steady_clock::now()
        + std::chrono::milliseconds(SBSTREAMER_STANDBY_TIMEOUT);
    for (;;) {
        bool obsolete = false, disconnect = false;
        {
            std::lock_guard<std::mutex> lock(g_enc_mutex);
            obsolete = (unsigned)stream != get_squeezebox_stream_id();
            // An activated request must always take the active cleanup path,
            // even if its peer closes immediately after promotion.
            if (request->encoder) {
                enc = request->encoder;
                opened = request->opened;
                break;
            }
            if (peerClosed()) {
                printf("stream %d: GET #%llu standby closed by client\n", stream, request->id);
                disconnect = true;
            } else if (!obsolete && !request->pauseEnded && !(IsAborted() || handle->aborted())
                       && !yeney_is_paused() && activeRequest
                       && activeRequest->encoder->hasAudio()
                       && std::chrono::steady_clock::now() >= standbyDeadline) {
                printf("stream %d: GET #%llu standby timeout\n", stream, request->id);
                disconnect = true;
            }
            disconnect = disconnect || request->pauseEnded || (IsAborted() || handle->aborted());
            if (obsolete || disconnect)
                standbyRequests.erase(std::remove(standbyRequests.begin(), standbyRequests.end(), request),
                                      standbyRequests.end());
        }
        if (disconnect || obsolete) {
            if (obsolete && !disconnect) redirect();
            handle->disconnect();
            return;
        }
        usleep(5000);
    }

    // Only ACTIVE gets an encoder and headers. A held GET still means an
    // ACTIVE request waiting for PCM while LMS is paused, never a standby.
    bool waitForResumeAudio = yeney_is_paused() || request->heldResume;
    auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(SBSTREAMER_RESUME_TIMEOUT);
    auto waitingForAudio = [&] {
        return !enc->hasAudio() || (request->heldResume && !request->resumeAcknowledged);
    };
    while (opened && !(speakerRelinquished.load() && yeney_is_paused()) && waitForResumeAudio && waitingForAudio() && !enc->cancelled() && !enc->responseEnded() && !(IsAborted() || handle->aborted()) && !peerClosed()
           && (unsigned)stream == get_squeezebox_stream_id()
           && std::chrono::steady_clock::now() < deadline) {
        // A marked resume already sent LMS play. Its q/s reply may clear the
        // transport state, but must not renew the hold or send another play.
        if (!request->heldResume) ResumeSqueezeBox(stream);
        usleep(10000);
    }
    char buf[SBSTREAMER_CHUNK];
    int r = 0;
    const bool heldResume = request->heldResume;
    const bool closedResume = heldResume && peerClosed();
    const bool expiredResume = heldResume && waitingForAudio()
        && std::chrono::steady_clock::now() >= deadline;
    if (opened && !(speakerRelinquished.load() && yeney_is_paused()) && !closedResume && (!waitForResumeAudio || !waitingForAudio()) && !enc->cancelled())
        r = enc->read(buf, sizeof(buf), SBSTREAMER_HTTP_IDLE_TIMEOUT, false, peerClosed);
    bool streamReady = r >= 4 && memcmp(buf, "fLaC", 4) == 0;
    const std::string streamingHeaders = "HTTP/1.1 200 OK\r\nServer: " + handle->serverName() + "\r\nConnection: close\r\n"
        "Content-Type: audio/flac\r\nTransfer-Encoding: chunked\r\n\r\n";
    if (speakerRelinquished.load() && yeney_is_paused()) {
        const std::string paused = "HTTP/1.1 200 OK\r\nContent-Type: audio/flac\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
        handle->send(paused.c_str(), paused.size());
    } else if (closedResume) {
        printf("held resume GET #%llu closed by client\n", request->id);
    } else if (!streamReady && waitForResumeAudio && (unsigned)stream != get_squeezebox_stream_id()) {
        if (heldResume)
            printf("held resume GET #%llu -> 302 stream %u\n", request->id, get_squeezebox_stream_id());
        // LMS play after stop can start a new delivery generation. The held
        // GET follows that URL instead of reporting an audio failure.
        redirect();
    } else if (!streamReady) {
        if (expiredResume) printf("held resume GET #%llu expired\n", request->id);
        printf("stream %d: no audio before timeout or connection replaced\n", stream);
        std::string error = "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
        handle->send(error.c_str(), error.size());
    } else {
        request->serving = true;
        if (heldResume) printf("held resume GET #%llu fed\n", request->id);
        printf("stream %d: serving current generation with fresh FLAC header\n", stream);
        diagnostics.encoder([enc] { return enc->encodedAudioMs(); });
        const auto responseStarted = std::chrono::steady_clock::now();
        ConsumptionDiagnostic consumption(*handle, playerName().trueName, "stream");
        size_t sent = 0;
        bool sendFailed = false;
        auto send = [&](const char* data, size_t size) {
            if (sendFailed) return false;
            errno = 0;
            if (handle->send(data, size)) {
                sent += size;
                return true;
            }
            const int error = errno; // capture before logging or cleanup changes it
            sendFailed = true;
            const double seconds = std::chrono::duration<double>(
                std::chrono::steady_clock::now() - responseStarted).count();
            if (deferCloseLog && (error == EPIPE || error == ECONNRESET)) {
                closeLog.failed(stream, seconds, sent, error);
                return false;
            }
            bool expected = false;
            {
                std::lock_guard<std::mutex> lock(g_enc_mutex);
                expected = (error == EPIPE || error == ECONNRESET)
                    && request->expectedCloseAt != std::chrono::steady_clock::time_point{}
                    && std::chrono::steady_clock::now() - request->expectedCloseAt < std::chrono::seconds(2);
            }
            if (expected)
                printf("stream %d: client closed after %.1f s (%zu bytes confirmed sent; %s; errno=%d)\n",
                    stream, seconds, sent, strerror(error), error);
            else printf("stream %d: send to Sonos failed after %.1f s (%s; errno=%d; "
                   "%zu bytes confirmed sent, partial failed write uncounted)\n",
                stream, seconds, error ? strerror(error) : "short write or closed socket",
                error, sent);
            return false;
        };
        auto sendChunk = [&](const char* data, size_t size) {
            char header[16];
            int length = snprintf(header, sizeof(header), "%x\r\n", (unsigned)size);
            return send(header, length) && send(data, size) && send("\r\n", 2);
        };
        if (send(streamingHeaders.c_str(), streamingHeaders.size()) && sendChunk(buf, r)) {
            while (!(IsAborted() || handle->aborted()) && (r = enc->read(buf, sizeof(buf), SBSTREAMER_HTTP_IDLE_TIMEOUT, false, peerClosed)) > 0) {
                if (!sendChunk(buf, r)) break;
            }
            send("0\r\n\r\n", 5);
        }
    }
    if (peerClosed()) printf("stream %d: client closed connection\n", stream);
    {
        std::lock_guard<std::mutex> lock(g_enc_mutex);
        const bool pauseEnded = enc->responseEnded();
        const bool retain = enc == canonicalEncoder && enc->hasAudio() && !yeney_is_paused()
            && !enc->producerRetired() && (unsigned)stream == get_squeezebox_stream_id();
        if (!pauseEnded && !retain) enc->cancel();
        if (activeRequest == request) {
            activeRequest.reset();
            if (!pauseEnded) {
                if (g_enc == enc && !retain) g_enc.reset();
                if ((unsigned)stream == get_squeezebox_stream_id() && !standbyRequests.empty()) {
                    auto newest = std::max_element(standbyRequests.begin(), standbyRequests.end(),
                        [](const std::shared_ptr<StreamRequest>& a, const std::shared_ptr<StreamRequest>& b) {
                            return a->id < b->id;
                        });
                    auto promoted = *newest;
                    standbyRequests.erase(newest);
                    activateRequest(promoted, true);
                }
            }
        }
    }
    // Preserve paused encoders and send EOF before disconnecting.
    handle->disconnect();
    if (enc->cancelled() && !enc->responseEnded()) enc->close();
    printf("stream %d: done\n", stream);

    printf("Done serving stream %d to Sonos\n", stream);
}