diff --git a/Makefile b/Makefile index 62d2453..51adac1 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,8 @@ CLI_SOURCES := Sources/CLI/TerebiCLI.swift Sources/CLI/PlayerCommand.swift Sourc SWIFT_SOURCES += Sources/App/StreamResolver.swift SWIFT_SOURCES += Sources/App/MiruroMetadata.swift CLI_SOURCES += Sources/App/StreamResolver.swift +SWIFT_SOURCES += Sources/App/StreamQueue.swift +CLI_SOURCES += Sources/App/StreamQueue.swift CONFIG_BUILD := .build/configuration CONFIG_BIN := $(shell xcrun swift build --scratch-path $(CONFIG_BUILD) -c release --show-bin-path) CONFIG_LIBRARY := $(CONFIG_BIN)/libTerebiConfiguration.a @@ -277,8 +279,8 @@ $(BUILD_DIR)/progress-reporting-check: Checks/progress-reporting.mm Sources/Core check-reporting: $(BUILD_DIR)/progress-reporting-check $(BUILD_DIR)/progress-reporting-check -$(BUILD_DIR)/player-command-check: Checks/player-command.swift Sources/CLI/PlayerCommand.swift Sources/App/YouTubeResolver.swift Sources/App/ResolvedSource.swift Sources/App/StreamResolver.swift Makefile | $(BUILD_DIR) check-dependencies - $(SWIFT) -target $(ARCH)-apple-macosx15.0 -swift-version 6 -parse-as-library -O $(DEBUG_FLAGS) Checks/player-command.swift Sources/CLI/PlayerCommand.swift Sources/App/YouTubeResolver.swift Sources/App/ResolvedSource.swift Sources/App/StreamResolver.swift -framework Foundation -o $@ +$(BUILD_DIR)/player-command-check: Checks/player-command.swift Sources/CLI/PlayerCommand.swift Sources/App/YouTubeResolver.swift Sources/App/ResolvedSource.swift Sources/App/StreamResolver.swift Sources/App/StreamQueue.swift Makefile | $(BUILD_DIR) check-dependencies + $(SWIFT) -target $(ARCH)-apple-macosx15.0 -swift-version 6 -parse-as-library -O $(DEBUG_FLAGS) Checks/player-command.swift Sources/CLI/PlayerCommand.swift Sources/App/YouTubeResolver.swift Sources/App/ResolvedSource.swift Sources/App/StreamResolver.swift Sources/App/StreamQueue.swift -framework Foundation -o $@ check-cli: $(BUILD_DIR)/player-command-check $(BUILD_DIR)/player-command-check diff --git a/Sources/App/PlaybackQueue.swift b/Sources/App/PlaybackQueue.swift index c738bfa..e648e4f 100644 --- a/Sources/App/PlaybackQueue.swift +++ b/Sources/App/PlaybackQueue.swift @@ -4,11 +4,14 @@ struct PlaybackQueue { struct Item: Identifiable { let id = UUID() let url: URL + var label: String? = nil + var displayName: String { label ?? url.lastPathComponent } } private(set) var items: [Item] = [] private(set) var currentID: UUID? - var currentURL: URL? { items.first { $0.id == currentID }?.url } + var currentItem: Item? { items.first { $0.id == currentID } } + var currentURL: URL? { currentItem?.url } private var currentIndex: Int? { items.firstIndex { $0.id == currentID } } var canPrevious: Bool { currentIndex.map { $0 > 0 } ?? false } var canNext: Bool { currentIndex.map { $0 + 1 < items.count } ?? false } @@ -18,6 +21,17 @@ struct PlaybackQueue { currentID = items.first?.id } + // Unlike file batches, stream entries already have an intentional playback order. + mutating func replaceStreams(_ entries: [(url: URL, label: String)]) { + items = entries.map { Item(url: $0.url, label: $0.label) } + currentID = items.first?.id + } + + mutating func updateCurrentLabel(_ label: String) { + guard let index = currentIndex else { return } + items[index].label = label + } + mutating func append(_ urls: [URL]) { items.append(contentsOf: sortedFiles(urls).map { Item(url: $0) }) } diff --git a/Sources/App/StreamQueue.swift b/Sources/App/StreamQueue.swift new file mode 100644 index 0000000..be9b138 --- /dev/null +++ b/Sources/App/StreamQueue.swift @@ -0,0 +1,69 @@ +import Foundation +import Darwin + +struct StreamQueue: Sendable { + struct Entry: Sendable { + let url: URL + let identity: String + let progressFile: URL + let label: String + } + let entries: [Entry] + + private struct Document: Decodable { + struct Item: Decodable { let url: String; let progress_file: String } + let version: Int + let items: [Item] + } + private struct InvalidQueue: LocalizedError { + var errorDescription: String? { "Invalid stream queue: expected version 1, 1–4096 unique supported videos and new absolute report files in private owned directories." } + } + + static func load(from file: URL) throws -> StreamQueue { + do { + guard file.isFileURL, file.path.hasPrefix("/") else { throw InvalidQueue() } + let descriptor = Darwin.open(file.path, O_RDONLY | O_NONBLOCK | O_NOFOLLOW | O_CLOEXEC) + guard descriptor >= 0 else { throw InvalidQueue() } + let handle = FileHandle(fileDescriptor: descriptor, closeOnDealloc: true) + defer { try? handle.close() } + var info = stat() + guard fstat(descriptor, &info) == 0, (info.st_mode & S_IFMT) == S_IFREG else { throw InvalidQueue() } + let data = try handle.read(upToCount: 8 * 1024 * 1024 + 1) ?? Data() + guard data.count <= 8 * 1024 * 1024 else { throw InvalidQueue() } + let document = try JSONDecoder().decode(Document.self, from: data) + guard document.version == 1, (1...4096).contains(document.items.count) else { throw InvalidQueue() } + var identities = Set() + var reports = Set() + let entries = try document.items.map { item -> Entry in + let stream = try StreamURL.parse(item.url) + let url: URL + let identity: String + let label: String + switch stream { + case .youtube(let canonical): + url = canonical + identity = canonical.absoluteString + label = "YouTube · " + (URLComponents(url: canonical, resolvingAgainstBaseURL: false)?.queryItems?.first?.value ?? "Video") + case .miruro(let mediaID, let episode): + url = URL(string: item.url)! + identity = "miruro:\(mediaID):\(episode)" + label = "\(url.lastPathComponent.replacingOccurrences(of: "-", with: " ")) · Episode \(episode)" + } + guard identities.insert(identity).inserted, + item.progress_file.hasPrefix("/"), !item.progress_file.contains("\0") else { throw InvalidQueue() } + let report = URL(fileURLWithPath: item.progress_file).standardizedFileURL + let parent = report.deletingLastPathComponent() + var directoryInfo = stat() + guard lstat(parent.path, &directoryInfo) == 0, + (directoryInfo.st_mode & S_IFMT) == S_IFDIR, + directoryInfo.st_uid == geteuid(), (directoryInfo.st_mode & 0o777) == 0o700, + parent.resolvingSymlinksInPath().path == parent.path, + reports.insert(report.path).inserted else { throw InvalidQueue() } + var existing = stat() + guard lstat(report.path, &existing) != 0, errno == ENOENT else { throw InvalidQueue() } + return Entry(url: url, identity: identity, progressFile: report, label: label) + } + return StreamQueue(entries: entries) + } catch { throw InvalidQueue() } + } +} diff --git a/Sources/CLI/PlayerCommand.swift b/Sources/CLI/PlayerCommand.swift index 190bddf..d8d11ea 100644 --- a/Sources/CLI/PlayerCommand.swift +++ b/Sources/CLI/PlayerCommand.swift @@ -5,12 +5,14 @@ enum PlayerCommand { case printDefaultConfig case launch(URL?, newInstance: Bool, noConfig: Bool, progressFile: URL?) case launchSource(sourceFile: URL, newInstance: Bool, noConfig: Bool, progressFile: URL?) + case launchQueue(queueFile: URL, newInstance: Bool, noConfig: Bool) static let usage = """ Usage: terebi [--new-instance] [--no-config] [FILE | STREAM_URL] terebi [--new-instance] [--no-config] --url STREAM_URL terebi --new-instance --url STREAM_URL --progress-file PATH terebi --new-instance --source-file PATH [--progress-file PATH] + terebi --new-instance [--no-config] --queue-file PATH terebi [--new-instance] [--no-config] -- FILE terebi --help terebi --default-config @@ -23,6 +25,7 @@ enum PlayerCommand { --no-config Use built-in settings instead of config.toml on a fresh launch. --url URL Open a YouTube video or Miruro episode URL. --source-file PATH Open a validated resolved HTTP(S) source in a new player. + --queue-file PATH Open an ordered version-1 stream queue with per-item reports. --progress-file PATH Write JSONL playback facts to a new private file; requires --new-instance. The GUI owns reporting after this CLI exits. -- Treat the following argument as a filename, even with a leading '-'. @@ -90,6 +93,15 @@ enum PlayerCommand { guard progressFile == nil else { throw ArgumentError("--progress-file cannot be combined with --default-config.") } return .printDefaultConfig } + if inputs.first == "--queue-file" { + guard newInstance, progressFile == nil, inputs.count == 2, + inputs[1].hasPrefix("/"), !inputs[1].contains("\0") else { + throw ArgumentError("--queue-file requires --new-instance and one absolute manifest path; it cannot be combined with --progress-file.") + } + let queueFile = URL(fileURLWithPath: inputs[1]).standardizedFileURL + _ = try StreamQueue.load(from: queueFile) + return .launchQueue(queueFile: queueFile, newInstance: newInstance, noConfig: noConfig) + } if inputs.first == "--source-file" { guard newInstance else { throw ArgumentError("--source-file requires --new-instance.") } guard inputs.count == 2, !inputs[1].isEmpty, !inputs[1].contains("\0"), diff --git a/Sources/CLI/TerebiCLI.swift b/Sources/CLI/TerebiCLI.swift index 4a59f79..7d8fa16 100644 --- a/Sources/CLI/TerebiCLI.swift +++ b/Sources/CLI/TerebiCLI.swift @@ -21,10 +21,14 @@ enum TerebiCLI { await launch(url: url, sourceFile: nil, newInstance: newInstance, noConfig: noConfig, progressFile: progressFile) case .launchSource(let sourceFile, let newInstance, let noConfig, let progressFile): await launch(url: nil, sourceFile: sourceFile, newInstance: newInstance, noConfig: noConfig, progressFile: progressFile) + case .launchQueue(let queueFile, let newInstance, let noConfig): + await launch(url: nil, sourceFile: nil, newInstance: newInstance, noConfig: noConfig, + progressFile: nil, queueFile: queueFile) } } - private static func launch(url: URL?, sourceFile: URL?, newInstance: Bool, noConfig: Bool, progressFile: URL?) async { + private static func launch(url: URL?, sourceFile: URL?, newInstance: Bool, noConfig: Bool, + progressFile: URL?, queueFile: URL? = nil) async { if let progressFile { var existing = stat() guard lstat(progressFile.path, &existing) != 0 else { @@ -69,6 +73,9 @@ enum TerebiCLI { if let progressFile { configuration.arguments += ["--progress-file", progressFile.path] } + if let queueFile { + configuration.arguments += ["--queue-file", queueFile.path] + } if let sourceFile { configuration.arguments += ["--source-file", sourceFile.path] } diff --git a/Sources/Core/player.cpp b/Sources/Core/player.cpp index ac4c887..13929af 100644 --- a/Sources/Core/player.cpp +++ b/Sources/Core/player.cpp @@ -197,7 +197,7 @@ struct Player { PlayerStatus status = PLAYER_EMPTY; bool previewShown = false; std::array error{}; - std::unique_ptr reporter; + std::unique_ptr reporter; bool remoteReporting = false; std::thread worker; @@ -1432,7 +1432,7 @@ bool player_set_progress_file(Player *player, const char *path, char *error) { if (player->reporter || player->stopping.load(std::memory_order_relaxed) || player->state.status != PLAYER_EMPTY || !player->commands.empty()) throw std::runtime_error("Progress reporting must be initialized once, before opening media"); - player->reporter = std::make_unique(path); + player->reporter = std::make_unique(path); return true; } catch (const std::exception &failure) { if (error) copyError(error, failure.what()); @@ -1441,6 +1441,24 @@ bool player_set_progress_file(Player *player, const char *path, char *error) { } return false; } +bool player_register_progress_file(Player *player, const char *identity, const char *path, char *error) { + if (error) error[0] = '\0'; + try { + if (!player || !identity || !path) throw std::runtime_error("Queue progress identity and path are required"); + std::lock_guard lock(player->mutex); + if (player->stopping.load(std::memory_order_relaxed) || + player->state.status != PLAYER_EMPTY || !player->commands.empty()) + throw std::runtime_error("Queue reports must be registered before opening media"); + if (!player->reporter) player->reporter = std::make_unique(); + player->reporter->registerFile(identity, path); + return true; + } catch (const std::exception &failure) { + if (error) copyError(error, failure.what()); + } catch (...) { + if (error) copyError(error, "Could not register queue progress reporting"); + } + return false; +} bool player_get_progress_error(Player *player, char *error) { if (error) error[0] = '\0'; if (!player || !player->reporter) return false; diff --git a/Sources/Core/player.h b/Sources/Core/player.h index b4a7c0c..f909993 100644 --- a/Sources/Core/player.h +++ b/Sources/Core/player.h @@ -40,6 +40,9 @@ Player *player_create(void); // Opt-in reporting must be initialized before opening media. Creates a new 0600 // file exclusively; errors never change playback state. Error buffers hold 512 bytes. bool player_set_progress_file(Player *player, const char *path, char *error); +// Register immutable queue destinations before opening any media. Each path is new; +// revisits append only to the original owned inode, and unmatched identities are not reported. +bool player_register_progress_file(Player *player, const char *identity, const char *path, char *error); bool player_get_progress_error(Player *player, char *error); // Stops the worker and drains final reporting, allowing errors to be shown before // destroying the handle. Idempotent; do not call from an audio callback. diff --git a/Sources/Core/progress_reporter.cpp b/Sources/Core/progress_reporter.cpp index 5dad1ce..a78b807 100644 --- a/Sources/Core/progress_reporter.cpp +++ b/Sources/Core/progress_reporter.cpp @@ -28,7 +28,7 @@ void appendNumber(std::string &record, double value) { } ProgressReporter::ProgressReporter(const std::string &path, std::function clock) - : clock(clock ? std::move(clock) : monotonicSeconds) { + : 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); @@ -49,6 +49,8 @@ ProgressReporter::ProgressReporter(const std::string &path, std::function(path)), active(single.get()) {} + +void ProgressReports::registerFile(const std::string &identity, const std::string &path) { + 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"); + for (const auto &[key, entry] : entries) + if (entry.path == path) 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 {}; +} } diff --git a/Sources/Core/progress_reporter.hpp b/Sources/Core/progress_reporter.hpp index 2849f6d..946fddd 100644 --- a/Sources/Core/progress_reporter.hpp +++ b/Sources/Core/progress_reporter.hpp @@ -7,6 +7,9 @@ #include #include #include +#include +#include +#include #include #include @@ -26,6 +29,8 @@ public: void closeMedia() noexcept; // Call after the decoder joins; drains queued records and joins the writer. void finish() noexcept; + // Reopen only this session's original private inode, never an arbitrary existing file. + void resume(); std::string error() const; private: @@ -36,6 +41,9 @@ private: bool eligible = false; } snapshot; std::string mediaURL; + std::string path; + dev_t device = 0; + ino_t inode = 0; std::function clock; bool live = false; bool hasSnapshot = false; @@ -58,4 +66,24 @@ private: void disable(const char *message, int osError = 0) noexcept; void writeRecords() noexcept; }; + +// Routes decoder-owned snapshots by immutable media identity; only the active writer runs. +class ProgressReports { +public: + explicit ProgressReports(const std::string &path); + ProgressReports() = default; + void registerFile(const std::string &identity, const std::string &path); + void begin(const std::string &identity, bool live = false) noexcept; + void observe(const PlayerState &state) noexcept { if (active) active->observe(state); } + void closeMedia() noexcept { if (active) active->closeMedia(); } + void finish() noexcept; + std::string error() const; +private: + struct Entry { std::string path; std::unique_ptr reporter; }; + std::unique_ptr single; + std::map entries; + ProgressReporter *active = nullptr; + mutable std::mutex mutex; + std::string failure; +}; }