// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
#include "queue_item_store.h"
#include "pcm_pack.h"
#include "wav_writer.h"
#include <FLAC/stream_decoder.h>
#include <FLAC/stream_encoder.h>
#include <algorithm>
#include <condition_variable>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <fcntl.h>
#include <fstream>
#include <map>
#include <mutex>
#include <sys/file.h>
#include <unistd.h>
#include <vector>

namespace queue_audio {
namespace {
constexpr size_t chunkFrames = 4096, frameBytes = 6;
struct MemoryFile {
    std::vector<uint8_t> bytes;
    size_t cursor = 0;
    bool failed = false;
};
FLAC__StreamEncoderWriteStatus encoded(const FLAC__StreamEncoder *, const FLAC__byte *p, size_t n, unsigned,
                                       unsigned, void *opaque) {
    auto &m = *static_cast<MemoryFile *>(opaque);
    try {
        if (m.cursor + n > m.bytes.size())
            m.bytes.resize(m.cursor + n);
        std::memcpy(m.bytes.data() + m.cursor, p, n);
        m.cursor += n;
        return FLAC__STREAM_ENCODER_WRITE_STATUS_OK;
    } catch (...) {
        m.failed = true;
        return FLAC__STREAM_ENCODER_WRITE_STATUS_FATAL_ERROR;
    }
}
FLAC__StreamEncoderSeekStatus seekEncoded(const FLAC__StreamEncoder *, FLAC__uint64 pos, void *opaque) {
    auto &m = *static_cast<MemoryFile *>(opaque);
    if (pos > m.bytes.size())
        return FLAC__STREAM_ENCODER_SEEK_STATUS_ERROR;
    m.cursor = pos;
    return FLAC__STREAM_ENCODER_SEEK_STATUS_OK;
}
FLAC__StreamEncoderTellStatus tellEncoded(const FLAC__StreamEncoder *, FLAC__uint64 *pos, void *opaque) {
    *pos = static_cast<MemoryFile *>(opaque)->cursor;
    return FLAC__STREAM_ENCODER_TELL_STATUS_OK;
}
std::vector<uint8_t> compress(const std::vector<char> &pcm, uint32_t rate) {
    MemoryFile memory;
    auto encoder = std::unique_ptr<FLAC__StreamEncoder, decltype(&FLAC__stream_encoder_delete)>(
        FLAC__stream_encoder_new(), FLAC__stream_encoder_delete);
    if (!encoder)
        throw std::bad_alloc();
    auto *e = encoder.get();
    bool ok =
        FLAC__stream_encoder_set_verify(e, true) && FLAC__stream_encoder_set_channels(e, 2) &&
        FLAC__stream_encoder_set_bits_per_sample(e, 24) && FLAC__stream_encoder_set_sample_rate(e, rate) &&
        FLAC__stream_encoder_set_compression_level(e, 5) && FLAC__stream_encoder_set_blocksize(e, 4096) &&
        FLAC__stream_encoder_set_do_mid_side_stereo(e, false) &&
        FLAC__stream_encoder_set_total_samples_estimate(e, pcm.size() / frameBytes);
    if (!ok || FLAC__stream_encoder_init_stream(e, encoded, seekEncoded, tellEncoded, nullptr, &memory) !=
                   FLAC__STREAM_ENCODER_INIT_STATUS_OK)
        throw std::runtime_error("history FLAC init failed");
    std::vector<FLAC__int32> samples(pcm.size() / 3);
    for (size_t i = 0; i < samples.size(); ++i) {
        const auto *p = reinterpret_cast<const unsigned char *>(pcm.data() + i * 3);
        uint32_t n = uint32_t(p[0]) | uint32_t(p[1]) << 8 | uint32_t(p[2]) << 16;
        samples[i] = int32_t(n & 0x800000 ? int64_t(n) - 0x1000000 : n);
    }
    ok = FLAC__stream_encoder_process_interleaved(e, samples.data(), pcm.size() / frameBytes);
    ok = FLAC__stream_encoder_finish(e) && ok;
    if (!ok || memory.failed)
        throw std::runtime_error("history FLAC encode failed");
    return std::move(memory.bytes);
}
struct Decoded {
    std::vector<char> bytes;
    size_t expected;
    uint32_t rate;
    bool failed = false;
};
FLAC__StreamDecoderWriteStatus decoded(const FLAC__StreamDecoder *, const FLAC__Frame *f,
                                       const FLAC__int32 *const channels[], void *opaque) {
    auto &d = *static_cast<Decoded *>(opaque);
    if (f->header.channels != 2 || f->header.bits_per_sample != 24 || f->header.sample_rate != d.rate ||
        f->header.blocksize * frameBytes > d.expected - d.bytes.size()) {
        d.failed = true;
        return FLAC__STREAM_DECODER_WRITE_STATUS_ABORT;
    }
    try {
        for (unsigned i = 0; i < f->header.blocksize; ++i)
            for (unsigned c = 0; c < 2; ++c)
                for (unsigned b = 0; b < 3; ++b)
                    d.bytes.push_back(char(uint32_t(channels[c][i]) >> (8 * b)));
    } catch (...) {
        d.failed = true;
        return FLAC__STREAM_DECODER_WRITE_STATUS_ABORT;
    }
    return FLAC__STREAM_DECODER_WRITE_STATUS_CONTINUE;
}
void decodeError(const FLAC__StreamDecoder *, FLAC__StreamDecoderErrorStatus, void *opaque) {
    static_cast<Decoded *>(opaque)->failed = true;
}
std::vector<char> decompress(FILE *file, size_t frames, uint32_t rate) {
    Decoded data{{}, frames * frameBytes, rate};
    data.bytes.reserve(data.expected);
    auto decoder = std::unique_ptr<FLAC__StreamDecoder, decltype(&FLAC__stream_decoder_delete)>(
        FLAC__stream_decoder_new(), FLAC__stream_decoder_delete);
    if (!decoder) {
        fclose(file);
        throw std::bad_alloc();
    }
    FLAC__stream_decoder_set_md5_checking(decoder.get(), true);
    if (FLAC__stream_decoder_init_FILE(decoder.get(), file, decoded, nullptr, decodeError, &data) !=
        FLAC__STREAM_DECODER_INIT_STATUS_OK)
        throw std::runtime_error("history FLAC open failed");
    bool ok = FLAC__stream_decoder_process_until_end_of_stream(decoder.get());
    ok = FLAC__stream_decoder_finish(decoder.get()) && ok;
    if (!ok || data.failed || data.bytes.size() != data.expected)
        throw std::runtime_error("history FLAC corrupt");
    return std::move(data.bytes);
}
} // namespace
LengthFault::LengthFault(uint64_t e, uint64_t a)
    : std::runtime_error("item length fault expected=" + std::to_string(e) + " actual=" + std::to_string(a)),
      expected(e), actual(a) {}
Config Config::environment() {
    Config c;
    if (const char *directory = std::getenv("STATE_DIRECTORY")) {
        std::filesystem::path root(directory);
        if (!root.is_absolute() || std::string(directory).find(':') != std::string::npos)
            throw std::invalid_argument("invalid queue STATE_DIRECTORY");
        c.root = root / "queue";
    }
    if (const char *p = std::getenv("YENEY_QUEUE_DISK_MIB")) {
        std::string s(p);
        size_t used = 0;
        if (s.empty() || s.find_first_not_of("0123456789") != std::string::npos)
            throw std::invalid_argument("invalid YENEY_QUEUE_DISK_MIB");
        auto mib = std::stoull(s, &used);
        if (!mib || used != s.size() || mib > UINT64_MAX / (1024 * 1024))
            throw std::invalid_argument("invalid YENEY_QUEUE_DISK_MIB");
        c.diskBytes = mib * 1024 * 1024;
        c.estimatedDiskBytes = c.diskBytes;
    }
    return c;
}
struct RamChunk {
    uint64_t first;
    std::vector<char> bytes;
};
struct HistoryChunk {
    uint64_t first;
    size_t frames;
    std::filesystem::path path;
    uint64_t bytes;
};
struct ItemState {
    const Descriptor descriptor;
    const uint64_t serial;
    uint64_t produced = 0;
    bool finished = false, protectedItem = false, interrupted = false;
    std::exception_ptr fault;
    std::deque<RamChunk> ram;
    std::vector<HistoryChunk> history;
    ItemState(Descriptor d, uint64_t id) : descriptor(std::move(d)), serial(id) {}
};
struct StoreState {
    mutable std::mutex mutex;
    std::mutex producer; // one storage writer; never held by readers/cancellation
    std::condition_variable changed;
    Config config;
    std::filesystem::path directory;
    std::map<std::string, std::shared_ptr<ItemState>> items;
    uint64_t serial = 0, disk = 0, encodedReserved = 0;
    size_t ram = 0;
    bool stopped = false;
    int lockFd = -1;
    ~StoreState() {
        if (lockFd >= 0)
            close(lockFd);
    }
    void check(const ItemState &item) const {
        if (item.fault)
            std::rethrow_exception(item.fault);
        if (stopped)
            throw std::runtime_error("item store closed");
    }
    void fail(ItemState &item, std::exception_ptr fault) {
        item.fault = std::move(fault);
        for (const auto &chunk : item.ram)
            ram -= chunk.bytes.capacity();
        item.ram.clear();
        changed.notify_all();
    }
    void space(uint64_t amount, const ItemState *writing) {
        while (disk + encodedReserved > config.diskBytes || amount > config.diskBytes - disk - encodedReserved) {
            std::shared_ptr<ItemState> oldest;
            for (const auto &pair : items) {
                const auto &i = pair.second;
                if (i.get() != writing && (i->finished || i->fault) && !i->protectedItem &&
                    !i->history.empty() && (!oldest || i->serial < oldest->serial))
                    oldest = i;
            }
            if (!oldest)
                throw std::runtime_error("queue disk quota exhausted: current/next retained");
            for (const auto &h : oldest->history) {
                if (!std::filesystem::remove(h.path))
                    throw std::runtime_error("history eviction failed");
                disk -= h.bytes;
            }
            oldest->history.clear();
            oldest->fault =
                std::make_exception_ptr(std::runtime_error("item history evicted; LMS re-render required"));
            changed.notify_all();
        }
    }
    void spill(ItemState &item, std::unique_lock<std::mutex> &lock) {
        // Snapshot RAM under the metadata lock, then encode/write without it.
        // The single writer owns producer; cancellation may still wake readers.
        auto snapshot = item.ram.front();
        lock.unlock();
        std::vector<uint8_t> flac;
        try {
            flac = compress(snapshot.bytes, item.descriptor.rate);
        } catch (...) {
            lock.lock();
            throw;
        }
        lock.lock();
        check(item);
        auto &r = item.ram.front();
        space(flac.size(), &item);
        const auto path = directory / (std::to_string(item.serial) + "-" + std::to_string(r.first) + ".flac");
        // FLAC is fully encoded in bounded temporary RAM before quota admission.
        // No WAV or raw PCM file is ever written.
        if (item.history.size() == item.history.capacity())
            item.history.reserve(item.history.capacity() ? item.history.capacity() * 2 : 16);
        HistoryChunk record{r.first, r.bytes.size() / frameBytes, path, flac.size()};
        lock.unlock();
        std::ofstream file(path, std::ios::binary | std::ios::trunc);
        file.write(reinterpret_cast<const char *>(flac.data()), flac.size());
        file.close();
        if (!file) {
            lock.lock();
            std::filesystem::remove(path);
            throw std::runtime_error("history write failed");
        }
        lock.lock();
        try {
            check(item);
        } catch (...) {
            std::filesystem::remove(path);
            throw;
        }
        item.history.push_back(std::move(record));
        disk += flac.size();
        ram -= r.bytes.capacity();
        item.ram.pop_front();
    }
    void trimRam(size_t reserve, std::unique_lock<std::mutex> &lock) {
        while (reserve > config.ramBytes - ram) {
            std::shared_ptr<ItemState> oldest;
            for (const auto &pair : items)
                if (!pair.second->ram.empty() && (!oldest || pair.second->serial < oldest->serial))
                    oldest = pair.second;
            spill(*oldest, lock);
        }
    }
};
Store::Store(const std::string &room, Config c) : state(std::make_shared<StoreState>()) {
    if (room.empty() || room == "." || room == ".." || room.find('/') != std::string::npos ||
        room.find('\0') != std::string::npos || c.ramBytes < frameBytes || !c.diskBytes)
        throw std::invalid_argument("invalid queue store configuration");
    state->config = std::move(c);
    state->directory = state->config.root / room;
    if (std::filesystem::is_symlink(state->config.root) || std::filesystem::is_symlink(state->directory))
        throw std::runtime_error("queue storage directory is a symlink");
    std::filesystem::create_directories(state->directory);
    std::filesystem::permissions(state->directory, std::filesystem::perms::owner_all);
    state->lockFd =
        open((state->directory / ".lock").c_str(), O_RDWR | O_CREAT | O_CLOEXEC | O_NOFOLLOW, 0600);
    if (state->lockFd < 0 || flock(state->lockFd, LOCK_EX | LOCK_NB))
        throw std::runtime_error("queue room store already open or unavailable");
    for (const auto &entry : std::filesystem::directory_iterator(state->directory))
        if (entry.path().filename() != ".lock")
            std::filesystem::remove_all(entry.path());
}
Store::~Store() {
    std::lock_guard<std::mutex> lock(state->mutex);
    state->stopped = true;
    state->changed.notify_all();
    if (state->lockFd >= 0) {
        close(state->lockFd);
        state->lockFd = -1;
    }
}
std::shared_ptr<Item> Store::create(Descriptor descriptor) {
    if (descriptor.occurrence.empty() || descriptor.occurrence.find('\0') != std::string::npos ||
        !descriptor.rate || descriptor.rate > 655350 || descriptor.channels != 2 || descriptor.bits != 24)
        throw std::invalid_argument("invalid item descriptor");
    if (descriptor.kind == PromiseKind::UpperBound && !descriptor.promisedFrames)
        throw std::invalid_argument("upper-bound item needs declared frames");
    if (descriptor.promisedFrames)
        wavLength(*descriptor.promisedFrames);
    std::lock_guard<std::mutex> lock(state->mutex);
    const auto key = std::to_string(descriptor.generation) + ":" + descriptor.occurrence;
    if (state->items.count(key))
        throw std::invalid_argument("duplicate occurrence/generation");
    if (descriptor.kind == PromiseKind::UpperBound)
        state->config.diskBytes = std::max(state->config.diskBytes, state->config.estimatedDiskBytes);
    auto item = std::make_shared<ItemState>(std::move(descriptor), ++state->serial);
    state->items.emplace(key, item);
    return std::shared_ptr<Item>(new Item(state, item));
}
std::shared_ptr<Item> Store::publish(const std::shared_ptr<Item> &pending, uint64_t promised, PromiseKind kind) {
    wavLength(promised);
    std::lock_guard<std::mutex> writer(state->producer);
    std::lock_guard<std::mutex> lock(state->mutex);
    if (!pending || pending->store != state)
        throw std::invalid_argument("foreign pending item");
    auto &old = *pending->state;
    state->check(old);
    if (old.descriptor.promisedFrames || old.finished || old.produced > promised)
        throw std::invalid_argument("publication conflicts with produced frames");
    auto descriptor = old.descriptor;
    descriptor.promisedFrames = promised;
    descriptor.kind = kind;
    ++descriptor.generation;
    const auto key = std::to_string(descriptor.generation) + ":" + descriptor.occurrence;
    if (state->items.count(key))
        throw std::invalid_argument("duplicate publication generation");
    auto item = std::make_shared<ItemState>(descriptor, old.serial);
    item->produced = old.produced;
    item->ram = std::move(old.ram);
    item->history = std::move(old.history);
    item->protectedItem = old.protectedItem;
    old.fault = std::make_exception_ptr(PublicationSuperseded());
    state->items.emplace(key, item);
    state->changed.notify_all();
    return std::shared_ptr<Item>(new Item(state, item));
}
void Store::protect(const std::shared_ptr<Item> &current, const std::shared_ptr<Item> &next) {
    protect(std::vector<std::shared_ptr<Item>>{current, next});
}
void Store::protect(const std::vector<std::shared_ptr<Item>> &items) {
    std::lock_guard<std::mutex> lock(state->mutex);
    for (auto item : items)
        if (item && item->store != state)
            throw std::invalid_argument("foreign protected item");
    for (auto &pair : state->items)
        pair.second->protectedItem = false;
    for (auto item : items)
        if (item)
            item->state->protectedItem = true;
}
std::optional<yeney::EncodedSpool> Store::reserveEncoded(bool enabled) {
    std::lock_guard<std::mutex> lock(state->mutex);
    state->encodedReserved = 0;
    if (!enabled) return std::nullopt;
    state->config.diskBytes = std::max(state->config.diskBytes,state->config.estimatedDiskBytes);
    const auto bytes = std::min<uint64_t>(state->config.diskBytes / 2,UINT32_MAX-65536ull);
    if (!bytes) throw std::runtime_error("queue disk quota cannot reserve encoded input");
    return yeney::EncodedSpool{state->directory.string(),bytes,[held=state](uint64_t allocated) {
        std::lock_guard<std::mutex> lock(held->mutex);
        if (held->stopped) throw std::runtime_error("encoded spool store closed");
        if (allocated > held->encodedReserved) held->space(allocated-held->encodedReserved,nullptr);
        held->encodedReserved = allocated;
    }};
}
uint64_t Store::diskBytes() const {
    std::lock_guard<std::mutex> lock(state->mutex);
    return state->disk + state->encodedReserved;
}
size_t Store::ramBytes() const {
    std::lock_guard<std::mutex> lock(state->mutex);
    return state->ram;
}
Item::Item(std::shared_ptr<StoreState> s, std::shared_ptr<ItemState> i)
    : store(std::move(s)), state(std::move(i)) {}
const Descriptor &Item::descriptor() const { return state->descriptor; }
std::string Item::room() const { return store->directory.filename().string(); }
std::optional<uint64_t> Item::contentLength() const {
    std::lock_guard<std::mutex> lock(store->mutex);
    store->check(*state);
    if (state->descriptor.promisedFrames)
        return wavLength(*state->descriptor.promisedFrames);
    return state->finished ? std::optional<uint64_t>(wavLength(state->produced)) : std::nullopt;
}
bool Item::finished() const {
    std::lock_guard<std::mutex> lock(store->mutex);
    store->check(*state);
    return state->finished;
}
uint64_t Item::producedFrames() const {
    std::lock_guard<std::mutex> lock(store->mutex);
    return state->produced;
}
void Item::append(const yeney::Frame *frames, size_t count, bool processed) {
    std::lock_guard<std::mutex> writer(store->producer);
    std::unique_lock<std::mutex> lock(store->mutex);
    store->check(*state);
    if (state->finished || (!frames && count))
        throw std::invalid_argument("invalid item append");
    try {
        if (count > UINT64_MAX - state->produced)
            throw std::overflow_error("frame count overflow");
        auto actual = state->produced + count;
        if (state->descriptor.promisedFrames && actual > *state->descriptor.promisedFrames)
            throw LengthFault(*state->descriptor.promisedFrames, actual);
        wavLength(actual);
        PcmDither unused;
        while (count) {
            const size_t allocation = std::min(chunkFrames, store->config.ramBytes / frameBytes) * frameBytes;
            if (state->ram.empty() || state->ram.back().bytes.size() == allocation) {
                store->trimRam(allocation, lock);
                RamChunk chunk{state->produced, {}};
                chunk.bytes.reserve(allocation);
                state->ram.push_back(std::move(chunk));
                store->ram += state->ram.back().bytes.capacity();
            }
            auto &bytes = state->ram.back().bytes;
            const size_t n = std::min(count, (bytes.capacity() - bytes.size()) / frameBytes);
            const auto base = bytes.size();
            bytes.resize(base + n * frameBytes);
            for (size_t i = 0; i < n; ++i) {
                packSonosSample(bytes.data() + base + i * 6, frames[i].left, 3, processed, unused);
                packSonosSample(bytes.data() + base + i * 6 + 3, frames[i].right, 3, processed, unused);
            }
            state->produced += n;
            count -= n;
            frames += n;
        }
    } catch (...) {
        store->fail(*state, std::current_exception());
        throw;
    }
    store->changed.notify_all();
}
void Item::finish() {
    std::lock_guard<std::mutex> writer(store->producer);
    std::unique_lock<std::mutex> lock(store->mutex);
    store->check(*state);
    if (state->finished)
        return;
    try {
        if (state->descriptor.kind == PromiseKind::Exact && state->descriptor.promisedFrames &&
            state->produced != *state->descriptor.promisedFrames)
            throw LengthFault(*state->descriptor.promisedFrames, state->produced);
        while (!state->ram.empty())
            store->spill(*state, lock);
        state->finished = true;
    } catch (...) {
        store->fail(*state, std::current_exception());
        throw;
    }
    store->changed.notify_all();
}
void Item::interrupt() {
    std::lock_guard<std::mutex> writer(store->producer);
    std::unique_lock<std::mutex> lock(store->mutex);
    store->check(*state);
    if (state->finished) return;
    while (!state->ram.empty()) store->spill(*state, lock);
    state->interrupted = true;
    state->finished = true;
    store->changed.notify_all();
}
void Item::cancel(const std::string &reason) {
    std::lock_guard<std::mutex> lock(store->mutex);
    if (!state->fault)
        store->fail(*state, std::make_exception_ptr(ItemCancelled(reason)));
}
void Item::releaseAudio() {
    std::lock_guard<std::mutex> lock(store->mutex);
    store->fail(*state, std::make_exception_ptr(ItemCancelled("item left queue window")));
    while (!state->history.empty()) {
        const auto &chunk = state->history.back();
        std::filesystem::remove(chunk.path);
        store->disk -= chunk.bytes;
        state->history.pop_back();
    }
    state->protectedItem = false;
}
Reader Item::reader(uint64_t pos) const { return Reader(store, state, pos); }
Reader::Reader(std::shared_ptr<StoreState> s, std::shared_ptr<ItemState> i, uint64_t pos)
    : store(std::move(s)), state(std::move(i)), offset(pos) {}
size_t Reader::read(char *output, size_t capacity, std::chrono::milliseconds timeout) {
    if (!capacity)
        return 0;
    if (!output)
        throw std::invalid_argument("null read buffer");
    std::unique_lock<std::mutex> lock(store->mutex);
    const auto deadline = std::chrono::steady_clock::now() + timeout;
    for (;;) {
        store->check(*state);
        auto frames = state->descriptor.promisedFrames;
        if (!frames && state->finished)
            frames = state->produced;
        if (frames) {
            const auto length = wavLength(*frames);
            if (offset > length)
                throw std::out_of_range("range beyond item");
            if (offset == length)
                return 0;
            if (offset < 44) {
                const auto h = wavHeader(state->descriptor.rate, *frames);
                const auto n = std::min<uint64_t>(capacity, 44 - offset);
                std::memcpy(output, h.data() + offset, n);
                offset += n;
                return n;
            }
            const uint64_t pcmOffset = offset - 44;
            if (state->interrupted && pcmOffset >= state->produced * frameBytes)
                throw ItemCancelled("LMS interrupted production; retained prefix only");
            if (state->finished && state->descriptor.kind == PromiseKind::UpperBound &&
                pcmOffset >= state->produced * frameBytes)
                return 0;
            if (pcmOffset < state->produced * frameBytes) {
                auto history =
                    std::upper_bound(state->history.begin(), state->history.end(), pcmOffset / frameBytes,
                                     [](uint64_t frame, const HistoryChunk &h) { return frame < h.first; });
                if (history != state->history.begin()) {
                    const auto h = *--history;
                    if (pcmOffset < (h.first + h.frames) * frameBytes) {
                        if (decodedFirst != h.first) {
                            try {
                                // Open while indexed; an open inode survives concurrent
                                // eviction.
                                FILE *file = fopen(h.path.c_str(), "rb");
                                if (!file)
                                    throw std::runtime_error("history FLAC open failed");
                                const auto first = h.first;
                                const auto frames = h.frames;
                                lock.unlock();
                                try {
                                    decodedChunk = decompress(file, frames, state->descriptor.rate);
                                } catch (...) {
                                    lock.lock();
                                    throw;
                                }
                                lock.lock();
                                store->check(*state);
                                decodedFirst = first;
                            } catch (...) {
                                store->fail(*state, std::current_exception());
                                throw;
                            }
                            decodedFirst = h.first;
                        }
                        const auto start = pcmOffset - h.first * frameBytes;
                        const auto n = std::min<uint64_t>(capacity, decodedChunk.size() - start);
                        std::memcpy(output, decodedChunk.data() + start, n);
                        offset += n;
                        return n;
                    }
                }
                for (const auto &r : state->ram) {
                    if (pcmOffset >= r.first * frameBytes &&
                        pcmOffset < r.first * frameBytes + r.bytes.size()) {
                        const auto start = pcmOffset - r.first * frameBytes;
                        const auto n = std::min<uint64_t>(capacity, r.bytes.size() - start);
                        std::memcpy(output, r.bytes.data() + start, n);
                        offset += n;
                        return n;
                    }
                }
                throw std::runtime_error("missing produced history");
            }
        }
        if (store->changed.wait_until(lock, deadline) == std::cv_status::timeout) {
            store->check(*state);
            throw ReadTimeout();
        }
    }
}
} // namespace queue_audio