#include "progress_reporter.hpp" #include #include #include #include #include #include #include #include #include #include #include #include namespace playback { namespace { double monotonicSeconds() { return std::chrono::duration(std::chrono::steady_clock::now().time_since_epoch()).count(); } void appendNumber(std::string &record, double value) { char buffer[64]; auto result = std::to_chars(buffer, buffer + sizeof(buffer), value, std::chars_format::general, std::numeric_limits::max_digits10); if (result.ec != std::errc{}) throw std::runtime_error("Could not encode progress timestamp"); record.append(buffer, result.ptr); } } ProgressReporter::ProgressReporter(const std::string &path, std::function clock) : path(path), clock(clock ? std::move(clock) : monotonicSeconds) { if (path.empty() || path.find('\0') != std::string::npos) throw std::invalid_argument("The progress file path must be nonempty and contain no NUL bytes"); descriptor = ::open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600); if (descriptor < 0) { const int code = errno; throw std::runtime_error("Could not exclusively create progress file '" + path + "': " + std::strerror(code)); } try { struct stat info{}; if (::fstat(descriptor, &info) != 0) throw std::runtime_error(std::string("Could not inspect progress file: ") + std::strerror(errno)); if (!S_ISREG(info.st_mode) || info.st_uid != ::geteuid() || info.st_nlink != 1) throw std::runtime_error("The new progress file must be a regular file privately owned by this user"); // Creation can only remove owner permissions through umask; restore exactly 0600. if (::fchmod(descriptor, 0600) != 0) throw std::runtime_error(std::string("Could not make progress file private: ") + std::strerror(errno)); if (::fstat(descriptor, &info) != 0) throw std::runtime_error(std::string("Could not verify progress file permissions: ") + std::strerror(errno)); if ((info.st_mode & 07777) != 0600) throw std::runtime_error("The progress file must have exactly owner read/write permissions (0600)"); device = info.st_dev; inode = info.st_ino; writer = std::thread([this] { writeRecords(); }); } catch (...) { ::close(descriptor); descriptor = -1; throw; } } ProgressReporter::~ProgressReporter() { finish(); } void ProgressReporter::resume() { if (!finished) return; if (disabled.load(std::memory_order_acquire)) throw std::runtime_error("Cannot resume a failed progress report"); descriptor = ::open(path.c_str(), O_WRONLY | O_APPEND | O_NONBLOCK | O_NOFOLLOW | O_CLOEXEC); if (descriptor < 0) throw std::runtime_error("Could not reopen the owned progress report"); struct stat info{}; if (::fstat(descriptor, &info) != 0 || !S_ISREG(info.st_mode) || info.st_dev != device || info.st_ino != inode || info.st_uid != ::geteuid() || info.st_nlink != 1 || (info.st_mode & 07777) != 0600) { ::close(descriptor); descriptor = -1; throw std::runtime_error("The owned progress report was replaced or is no longer private"); } try { stopping = false; writer = std::thread([this] { writeRecords(); }); finished = false; } catch (...) { ::close(descriptor); descriptor = -1; throw; } } bool ProgressReporter::validMediaURL(std::string_view url) noexcept { constexpr std::string_view prefix = "https://www.youtube.com/watch?v="; if (url.size() != prefix.size() + 11 || url.substr(0, prefix.size()) != prefix) return false; for (char character : url.substr(prefix.size())) { if (!((character >= 'a' && character <= 'z') || (character >= 'A' && character <= 'Z') || (character >= '0' && character <= '9') || character == '_' || character == '-')) return false; } return true; } bool ProgressReporter::validMediaIdentity(std::string_view identity) noexcept { if (validMediaURL(identity)) return true; constexpr std::string_view prefix = "miruro:"; if (identity.substr(0, prefix.size()) != prefix) return false; const auto separator = identity.find(':', prefix.size()); if (separator == std::string_view::npos) return false; const auto id = identity.substr(prefix.size(), separator - prefix.size()); if (id.empty() || id.size() > 512) return false; for (char c : id) { if (!((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_')) return false; } const auto episode = identity.substr(separator + 1); if (episode.empty() || episode.front() < '1' || episode.front() > '9') return false; for (char c : episode) if (c < '0' || c > '9') return false; int value = 0; const auto parsed = std::from_chars(episode.data(), episode.data() + episode.size(), value); return parsed.ec == std::errc{} && parsed.ptr == episode.data() + episode.size() && value > 0; } void ProgressReporter::begin(const std::string &url, bool isLive) noexcept { if (finished || disabled.load(std::memory_order_acquire)) return; if (url.empty()) { closeMedia(); return; } if (!validMediaIdentity(url)) { disable("Progress media identity is invalid"); return; } try { if (mediaURL != url) { closeMedia(); mediaURL = url; } live = isLive; awaitingReady = true; seekPending = false; } catch (const std::exception &exception) { disable(exception.what()); } catch (...) { disable("Could not begin playback progress reporting"); } } void ProgressReporter::observe(const PlayerState &state) noexcept { if (finished || mediaURL.empty() || disabled.load(std::memory_order_acquire)) return; if (state.seek_pending) { seekPending = true; return; } if (state.status != PLAYER_PAUSED && state.status != PLAYER_PLAYING && state.status != PLAYER_ENDED) return; if (!std::isfinite(state.position) || state.position < 0) return; // A later bookmark reset must not replace the native end-of-stream snapshot. if (hasSnapshot && snapshot.status == PLAYER_ENDED && state.status == PLAYER_ENDED && !awaitingReady && !seekPending) return; const bool paused = hasSnapshot && snapshot.status == PLAYER_PLAYING && state.status == PLAYER_PAUSED; const double duration = std::isfinite(state.duration) && state.duration > 0 ? state.duration : 0; const bool eligible = !live && state.has_video && state.can_seek && duration > 0; const bool crossing = !thresholdReported && eligible && state.position / duration >= .95; const bool immediate = awaitingReady || seekPending || paused || crossing || state.status == PLAYER_ENDED; snapshot = {state.position == 0 ? 0 : state.position, duration, state.status, eligible}; hasSnapshot = true; awaitingReady = false; seekPending = false; if (!immediate && state.status != PLAYER_PLAYING) return; try { const double now = clock(); if (!std::isfinite(now)) { disable("The playback progress clock returned a non-finite value"); return; } if (immediate || now - lastRecordAt >= 5) { if (emit(state.status == PLAYER_ENDED ? "ended" : "progress")) { lastRecordAt = now; if (crossing) thresholdReported = true; } } } catch (const std::exception &exception) { disable(exception.what()); } catch (...) { disable("Could not sample the playback progress clock"); } } void ProgressReporter::closeMedia() noexcept { if (mediaURL.empty()) return; if (hasSnapshot && !disabled.load(std::memory_order_acquire)) emit("closed"); mediaURL.clear(); hasSnapshot = false; awaitingReady = false; seekPending = false; thresholdReported = false; lastRecordAt = 0; } bool ProgressReporter::emit(const char *event) noexcept { try { std::string record; record.reserve(256); const bool miruro = mediaURL.compare(0, 7, "miruro:") == 0; record += miruro ? "{\"version\":2,\"event\":\"" : "{\"version\":1,\"event\":\""; record += event; if (miruro) { const auto separator = mediaURL.find(':', 7); record += "\",\"media_identity\":{\"provider\":\"miruro\",\"media_id\":\""; record.append(mediaURL, 7, separator - 7); record += "\",\"episode\":"; record.append(mediaURL, separator + 1, std::string::npos); record += "},\"position_seconds\":"; } else { record += "\",\"media_url\":\""; record += mediaURL; record += "\",\"position_seconds\":"; } appendNumber(record, snapshot.position); record += ",\"duration_seconds\":"; if (snapshot.duration > 0) appendNumber(record, snapshot.duration); else record += "null"; record += ",\"completion_eligible\":"; record += snapshot.eligible ? "true}\n" : "false}\n"; { std::lock_guard lock(mutex); if (stopping || disabled.load(std::memory_order_relaxed)) return false; records.push_back(std::move(record)); } wake.notify_one(); return true; } catch (const std::exception &exception) { disable(exception.what()); } catch (...) { disable("Could not encode or queue playback progress"); } return false; } void ProgressReporter::disable(const char *message, int osError) noexcept { std::lock_guard lock(mutex); if (!failure[0]) { if (osError) std::snprintf(failure.data(), failure.size(), "%s: %s", message, std::strerror(osError)); else std::snprintf(failure.data(), failure.size(), "%s", message); if (!failure[0]) std::snprintf(failure.data(), failure.size(), "Playback progress reporting failed"); } disabled.store(true, std::memory_order_release); records.clear(); wake.notify_all(); } std::string ProgressReporter::error() const { std::lock_guard lock(mutex); return failure.data(); } void ProgressReporter::writeRecords() noexcept { try { for (;;) { std::string record; { std::unique_lock lock(mutex); wake.wait(lock, [this] { return stopping || disabled.load(std::memory_order_relaxed) || !records.empty(); }); if (disabled.load(std::memory_order_relaxed) || (stopping && records.empty())) break; record = std::move(records.front()); records.pop_front(); } std::size_t written = 0; while (written < record.size() && !disabled.load(std::memory_order_acquire)) { const auto count = ::write(descriptor, record.data() + written, record.size() - written); if (count > 0) written += static_cast(count); else if (count < 0 && errno == EINTR) continue; else { const int code = count < 0 ? errno : EIO; disable("Could not write playback progress file", code); break; } } if (disabled.load(std::memory_order_acquire)) break; } } catch (const std::exception &exception) { disable(exception.what()); } catch (...) { disable("The playback progress writer failed"); } if (::close(descriptor) != 0) disable("Could not close playback progress file", errno); descriptor = -1; } void ProgressReporter::finish() noexcept { if (finished) return; closeMedia(); finished = true; { std::lock_guard lock(mutex); stopping = true; } wake.notify_all(); if (writer.joinable()) writer.join(); } ProgressReports::ProgressReports(const std::string &path) : single(std::make_unique(path)), active(single.get()) {} void ProgressReports::registerFile(const std::string &identity, const std::string &path) { std::lock_guard lock(mutex); if (single || !ProgressReporter::validMediaIdentity(identity) || entries.size() >= 4096 || path.empty() || path.front() != '/' || path.find('\0') != std::string::npos || entries.contains(identity)) throw std::runtime_error("Invalid or duplicate queue progress registration"); if (!paths.insert(path).second) throw std::runtime_error("Duplicate queue progress report path"); entries.emplace(identity, Entry{path, nullptr}); } void ProgressReports::begin(const std::string &identity, bool live) noexcept { std::lock_guard lock(mutex); if (single) { active->begin(identity, live); return; } const auto found = entries.find(identity); auto *next = found == entries.end() ? nullptr : found->second.reporter.get(); if (active != next && active) active->finish(); active = nullptr; // A failed or unregistered identity must never reuse the old report. if (found == entries.end()) return; try { auto &entry = found->second; if (!entry.reporter) entry.reporter = std::make_unique(entry.path); else entry.reporter->resume(); active = entry.reporter.get(); active->begin(identity, live); } catch (const std::exception &exception) { try { if (failure.empty()) failure = exception.what(); } catch (...) {} } catch (...) { try { if (failure.empty()) failure = "Could not select queue progress report"; } catch (...) {} } } void ProgressReports::finish() noexcept { std::lock_guard lock(mutex); if (active) active->finish(); } std::string ProgressReports::error() const { std::lock_guard lock(mutex); if (!failure.empty()) return failure; if (single) return single->error(); for (const auto &[identity, entry] : entries) { if (entry.reporter) { auto message = entry.reporter->error(); if (!message.empty()) return message; } } return {}; } }