feat: add saved-video collections and category views
This commit is contained in:
@@ -0,0 +1,974 @@
|
||||
use std::{
|
||||
collections::BTreeSet,
|
||||
fs::{self, File, OpenOptions, ReadDir},
|
||||
io::{self, Read, Write},
|
||||
path::{Path, PathBuf},
|
||||
sync::atomic::{AtomicU64, Ordering},
|
||||
};
|
||||
|
||||
use serde::Deserialize;
|
||||
use toml_edit::{Array, ArrayOfTables, DocumentMut, Table, value};
|
||||
|
||||
use crate::youtube::{Video, valid_video_id, validate_video};
|
||||
|
||||
const MAX_STATE_BYTES: u64 = 8 * 1024 * 1024;
|
||||
const MAX_LINE: usize = 16 * 1024;
|
||||
const STREAM_BYTES: usize = 16 * 1024;
|
||||
const MAX_STREAMS: usize = 4096;
|
||||
static NEXT_NAME: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
fn invalid(message: impl Into<String>) -> io::Error {
|
||||
io::Error::new(io::ErrorKind::InvalidData, message.into())
|
||||
}
|
||||
|
||||
struct StoredFile {
|
||||
path: PathBuf,
|
||||
original: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
impl StoredFile {
|
||||
fn open(path: PathBuf) -> io::Result<(Self, DocumentMut)> {
|
||||
let original = read_bytes(&path, MAX_STATE_BYTES)?;
|
||||
let document = match &original {
|
||||
Some(bytes) => std::str::from_utf8(bytes)
|
||||
.map_err(|e| invalid(format!("{} is not UTF-8: {e}", path.display())))?
|
||||
.parse::<DocumentMut>()
|
||||
.map_err(|e| {
|
||||
invalid(format!(
|
||||
"Invalid {}: {e}; repair this file before retrying",
|
||||
path.display()
|
||||
))
|
||||
})?,
|
||||
None => DocumentMut::new(),
|
||||
};
|
||||
if original.is_some() && document.get("version").and_then(|v| v.as_integer()) != Some(1) {
|
||||
return Err(invalid(format!("{} requires version = 1", path.display())));
|
||||
}
|
||||
Ok((Self { path, original }, document))
|
||||
}
|
||||
|
||||
fn save(&mut self, document: DocumentMut) -> io::Result<()> {
|
||||
let contents = document.to_string().into_bytes();
|
||||
atomic_write(&self.path, self.original.as_deref(), &contents)?;
|
||||
self.original = Some(contents);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
pub enum CollectionKind {
|
||||
WatchLater,
|
||||
Saved,
|
||||
}
|
||||
|
||||
struct Collection {
|
||||
file: StoredFile,
|
||||
videos: Vec<Video>,
|
||||
ids: BTreeSet<String>,
|
||||
}
|
||||
|
||||
impl Collection {
|
||||
fn open(path: PathBuf) -> io::Result<Self> {
|
||||
let (file, document) = StoredFile::open(path)?;
|
||||
let (videos, ids) =
|
||||
load_videos(&document).map_err(|e| invalid(format!("{}: {e}", file.path.display())))?;
|
||||
Ok(Self { file, videos, ids })
|
||||
}
|
||||
|
||||
fn save(&mut self, video: &Video) -> io::Result<()> {
|
||||
if self.ids.contains(&video.id) {
|
||||
return Ok(());
|
||||
}
|
||||
let mut video = video.clone();
|
||||
validate_saved(&mut video)?;
|
||||
self.file.save(video_document(
|
||||
self.videos.iter().chain(std::iter::once(&video)),
|
||||
))?;
|
||||
self.ids.insert(video.id.clone());
|
||||
self.videos.push(video);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remove(&mut self, ids: &[String]) -> io::Result<()> {
|
||||
if !ids.iter().any(|id| self.ids.contains(id)) {
|
||||
return Ok(());
|
||||
}
|
||||
self.file.save(video_document(
|
||||
self.videos.iter().filter(|v| !ids.contains(&v.id)),
|
||||
))?;
|
||||
self.videos.retain(|v| !ids.contains(&v.id));
|
||||
self.ids.retain(|id| !ids.contains(id));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub struct State {
|
||||
watch_later: Collection,
|
||||
saved_videos: Collection,
|
||||
watch_state: StoredFile,
|
||||
watched: BTreeSet<String>,
|
||||
pending: BTreeSet<String>,
|
||||
sessions: PathBuf,
|
||||
streams: Vec<ProgressStream>,
|
||||
discovery: Option<ReadDir>,
|
||||
next_stream: usize,
|
||||
notify_changed: bool,
|
||||
}
|
||||
|
||||
impl State {
|
||||
pub fn open(config_path: &Path) -> io::Result<Self> {
|
||||
let parent = config_path
|
||||
.parent()
|
||||
.filter(|p| !p.as_os_str().is_empty())
|
||||
.unwrap_or(Path::new("."));
|
||||
let watch_later = Collection::open(parent.join("watch-later.toml"))?;
|
||||
let saved_videos = Collection::open(parent.join("saved-videos.toml"))?;
|
||||
let (watch_state, document) = StoredFile::open(parent.join("watch-state.toml"))?;
|
||||
let watched = load_watched(&document)
|
||||
.map_err(|e| invalid(format!("{}: {e}", watch_state.path.display())))?;
|
||||
Ok(Self {
|
||||
watch_later,
|
||||
saved_videos,
|
||||
watch_state,
|
||||
watched,
|
||||
pending: BTreeSet::new(),
|
||||
sessions: parent.join("playback-sessions"),
|
||||
streams: Vec::new(),
|
||||
discovery: None,
|
||||
next_stream: 0,
|
||||
notify_changed: false,
|
||||
})
|
||||
}
|
||||
|
||||
fn collection(&self, kind: CollectionKind) -> &Collection {
|
||||
match kind {
|
||||
CollectionKind::WatchLater => &self.watch_later,
|
||||
CollectionKind::Saved => &self.saved_videos,
|
||||
}
|
||||
}
|
||||
|
||||
fn collection_mut(&mut self, kind: CollectionKind) -> &mut Collection {
|
||||
match kind {
|
||||
CollectionKind::WatchLater => &mut self.watch_later,
|
||||
CollectionKind::Saved => &mut self.saved_videos,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn videos(&self, kind: CollectionKind) -> &[Video] {
|
||||
&self.collection(kind).videos
|
||||
}
|
||||
pub fn contains(&self, kind: CollectionKind, id: &str) -> bool {
|
||||
self.collection(kind).ids.contains(id)
|
||||
}
|
||||
pub fn is_watched(&self, id: &str) -> bool {
|
||||
self.watched.contains(id)
|
||||
}
|
||||
|
||||
pub fn save(&mut self, kind: CollectionKind, video: &Video) -> io::Result<()> {
|
||||
self.collection_mut(kind).save(video)
|
||||
}
|
||||
|
||||
pub fn remove(&mut self, kind: CollectionKind, ids: &[String]) -> io::Result<()> {
|
||||
self.collection_mut(kind).remove(ids)
|
||||
}
|
||||
|
||||
pub fn progress_file(&mut self) -> io::Result<PathBuf> {
|
||||
if self.streams.len() >= MAX_STREAMS {
|
||||
return Err(invalid(
|
||||
"Playback session limit reached; remove inactive playback-sessions directories and restart",
|
||||
));
|
||||
}
|
||||
fs::create_dir_all(self.sessions.parent().unwrap_or(Path::new(".")))?;
|
||||
private_directory(&self.sessions, true)?;
|
||||
let directory = loop {
|
||||
let path = self.sessions.join(format!(
|
||||
"session-{}-{}",
|
||||
std::process::id(),
|
||||
NEXT_NAME.fetch_add(1, Ordering::Relaxed)
|
||||
));
|
||||
match create_private_directory(&path) {
|
||||
Ok(()) => break path,
|
||||
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
};
|
||||
let path = directory.join("progress.jsonl");
|
||||
self.streams.push(ProgressStream::new(path.clone()));
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
pub fn poll_progress(&mut self) -> io::Result<bool> {
|
||||
let mut error = self.discover().err();
|
||||
let count = self.streams.len().min(16);
|
||||
for _ in 0..count {
|
||||
if self.next_stream >= self.streams.len() {
|
||||
self.next_stream = 0;
|
||||
}
|
||||
let stream = &mut self.streams[self.next_stream];
|
||||
self.next_stream += 1;
|
||||
if let Err(e) = stream.poll(&mut self.pending, &self.watched)
|
||||
&& error.is_none()
|
||||
{
|
||||
error = Some(e);
|
||||
}
|
||||
}
|
||||
self.pending.retain(|id| !self.watched.contains(id));
|
||||
if !self.pending.is_empty() {
|
||||
self.watch_state.save(watched_document(
|
||||
self.watched.iter().chain(self.pending.iter()),
|
||||
))?;
|
||||
self.watched.append(&mut self.pending);
|
||||
self.notify_changed = true;
|
||||
}
|
||||
// Preserve notification after reporting a bad record; siblings were still processed.
|
||||
if let Some(error) = error {
|
||||
return Err(error);
|
||||
}
|
||||
Ok(std::mem::take(&mut self.notify_changed))
|
||||
}
|
||||
|
||||
fn discover(&mut self) -> io::Result<()> {
|
||||
match private_directory(&self.sessions, false) {
|
||||
Ok(()) => {}
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
if self.discovery.is_none() {
|
||||
self.discovery = Some(fs::read_dir(&self.sessions)?);
|
||||
}
|
||||
for _ in 0..32 {
|
||||
let Some(entry) = self.discovery.as_mut().unwrap().next() else {
|
||||
self.discovery = None;
|
||||
break;
|
||||
};
|
||||
let entry = entry?;
|
||||
if !entry.file_name().to_string_lossy().starts_with("session-") {
|
||||
continue;
|
||||
}
|
||||
let path = entry.path().join("progress.jsonl");
|
||||
if self.streams.iter().any(|stream| stream.path == path) {
|
||||
continue;
|
||||
}
|
||||
if private_directory(&entry.path(), false).is_err() {
|
||||
continue;
|
||||
}
|
||||
// Failed launchers leave empty private directories; only bound identities are discoverable.
|
||||
if !matches!(fs::symlink_metadata(entry.path().join("video-id")), Ok(m) if m.is_file())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if self.streams.len() >= MAX_STREAMS {
|
||||
break;
|
||||
}
|
||||
self.streams.push(ProgressStream::new(path));
|
||||
}
|
||||
// ponytail: retain reports for live GUI writers, capped at 4096 monitored sessions;
|
||||
// add explicit inactive-session archival if long-lived installations hit this ceiling.
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_saved(video: &mut Video) -> io::Result<()> {
|
||||
let date = video.publication_date.clone();
|
||||
if let Some(date) = &date {
|
||||
let raw = date.strip_prefix('~').unwrap_or(date);
|
||||
let compact = raw.replace('-', "");
|
||||
if !(raw.len() == 10 && raw.as_bytes()[4] == b'-' && raw.as_bytes()[7] == b'-')
|
||||
|| compact.len() != 8
|
||||
|| !compact.bytes().all(|b| b.is_ascii_digit())
|
||||
|| time::Date::from_calendar_date(
|
||||
compact[..4]
|
||||
.parse()
|
||||
.map_err(|_| invalid("Invalid saved publication date"))?,
|
||||
compact[4..6]
|
||||
.parse::<u8>()
|
||||
.ok()
|
||||
.and_then(|m| m.try_into().ok())
|
||||
.ok_or_else(|| invalid("Invalid saved publication month"))?,
|
||||
compact[6..]
|
||||
.parse()
|
||||
.map_err(|_| invalid("Invalid saved publication day"))?,
|
||||
)
|
||||
.is_err()
|
||||
{
|
||||
return Err(invalid(
|
||||
"Saved publication date must be a valid YYYY-MM-DD or ~YYYY-MM-DD",
|
||||
));
|
||||
}
|
||||
}
|
||||
validate_video(video).map_err(invalid)?;
|
||||
video.publication_date = date;
|
||||
if let Some(handle) = &video.uploader_id
|
||||
&& (!handle.starts_with('@') || crate::config::normalize_channel(handle).is_err())
|
||||
{
|
||||
return Err(invalid("Saved uploader_id must be a valid YouTube @handle"));
|
||||
}
|
||||
if video.channel_id.is_none() && video.uploader_id.is_none() {
|
||||
return Err(invalid(
|
||||
"Video has no channel identity; refresh its metadata before saving",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn load_videos(document: &DocumentMut) -> io::Result<(Vec<Video>, BTreeSet<String>)> {
|
||||
if document
|
||||
.iter()
|
||||
.any(|(key, _)| !matches!(key, "version" | "videos"))
|
||||
{
|
||||
return Err(invalid("Unknown watch-later field"));
|
||||
}
|
||||
let Some(item) = document.get("videos") else {
|
||||
return Ok((Vec::new(), BTreeSet::new()));
|
||||
};
|
||||
if item.as_array().is_some_and(|a| a.is_empty()) {
|
||||
return Ok((Vec::new(), BTreeSet::new()));
|
||||
}
|
||||
let tables = item
|
||||
.as_array_of_tables()
|
||||
.ok_or_else(|| invalid("videos must be an array of tables"))?;
|
||||
let mut videos = Vec::new();
|
||||
let mut ids = BTreeSet::new();
|
||||
for table in tables {
|
||||
let mut fields = serde_json::Map::new();
|
||||
for (key, item) in table.iter() {
|
||||
let field = match key {
|
||||
"id" | "title" | "channel" | "channel_id" | "uploader_id" | "upload_date" => {
|
||||
serde_json::Value::String(
|
||||
item.as_str()
|
||||
.ok_or_else(|| invalid(format!("Video {key} must be a string")))?
|
||||
.into(),
|
||||
)
|
||||
}
|
||||
"timestamp" => serde_json::Value::from(
|
||||
item.as_integer()
|
||||
.ok_or_else(|| invalid("Video timestamp must be an integer"))?,
|
||||
),
|
||||
_ => return Err(invalid(format!("Unknown saved video field {key:?}"))),
|
||||
};
|
||||
fields.insert(key.into(), field);
|
||||
}
|
||||
let mut video: Video = serde_json::from_value(serde_json::Value::Object(fields))
|
||||
.map_err(|e| invalid(format!("Invalid saved video: {e}")))?;
|
||||
validate_saved(&mut video)?;
|
||||
if ids.insert(video.id.clone()) {
|
||||
videos.push(video);
|
||||
}
|
||||
}
|
||||
Ok((videos, ids))
|
||||
}
|
||||
|
||||
fn load_watched(document: &DocumentMut) -> io::Result<BTreeSet<String>> {
|
||||
if document
|
||||
.iter()
|
||||
.any(|(key, _)| !matches!(key, "version" | "watched"))
|
||||
{
|
||||
return Err(invalid("Unknown watch-state field"));
|
||||
}
|
||||
let Some(item) = document.get("watched") else {
|
||||
return Ok(BTreeSet::new());
|
||||
};
|
||||
item.as_array()
|
||||
.ok_or_else(|| invalid("watched must be an array of YouTube IDs"))?
|
||||
.iter()
|
||||
.map(|v| {
|
||||
let id = v
|
||||
.as_str()
|
||||
.filter(|id| valid_video_id(id))
|
||||
.ok_or_else(|| invalid("Invalid watched YouTube ID"))?;
|
||||
Ok(id.to_owned())
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn video_document<'a>(videos: impl Iterator<Item = &'a Video>) -> DocumentMut {
|
||||
let mut document = DocumentMut::new();
|
||||
document["version"] = value(1);
|
||||
let mut tables = ArrayOfTables::new();
|
||||
for video in videos {
|
||||
let mut table = Table::new();
|
||||
table["id"] = value(video.id.as_str());
|
||||
table["title"] = value(video.title.as_str());
|
||||
for (key, field) in [
|
||||
("channel", &video.channel),
|
||||
("channel_id", &video.channel_id),
|
||||
("uploader_id", &video.uploader_id),
|
||||
("upload_date", &video.publication_date),
|
||||
] {
|
||||
if let Some(field) = field {
|
||||
table[key] = value(field.as_str());
|
||||
}
|
||||
}
|
||||
if let Some(timestamp) = video.timestamp {
|
||||
table["timestamp"] = value(timestamp);
|
||||
}
|
||||
tables.push(table);
|
||||
}
|
||||
document["videos"] = if tables.is_empty() {
|
||||
value(Array::new())
|
||||
} else {
|
||||
toml_edit::Item::ArrayOfTables(tables)
|
||||
};
|
||||
document
|
||||
}
|
||||
|
||||
fn watched_document<'a>(watched: impl Iterator<Item = &'a String>) -> DocumentMut {
|
||||
let mut document = DocumentMut::new();
|
||||
document["version"] = value(1);
|
||||
let mut ids = Array::new();
|
||||
for id in watched {
|
||||
ids.push(id.as_str());
|
||||
}
|
||||
document["watched"] = value(ids);
|
||||
document
|
||||
}
|
||||
|
||||
fn regular_file(path: &Path, private: bool) -> io::Result<File> {
|
||||
let mut options = OpenOptions::new();
|
||||
options.read(true);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
|
||||
}
|
||||
let file = options.open(path)?;
|
||||
let metadata = file.metadata()?;
|
||||
if !metadata.is_file() {
|
||||
return Err(invalid(format!(
|
||||
"{} must be a regular file",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
// SAFETY: geteuid has no preconditions.
|
||||
if metadata.uid() != unsafe { libc::geteuid() } || (private && metadata.mode() & 0o077 != 0)
|
||||
{
|
||||
return Err(invalid(format!(
|
||||
"{} must be owned by you with owner-only permissions",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(file)
|
||||
}
|
||||
|
||||
fn read_bytes(path: &Path, limit: u64) -> io::Result<Option<Vec<u8>>> {
|
||||
let file = match regular_file(path, false) {
|
||||
Ok(file) => file,
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
let mut bytes = Vec::new();
|
||||
file.take(limit + 1).read_to_end(&mut bytes)?;
|
||||
if bytes.len() as u64 > limit {
|
||||
return Err(invalid(format!(
|
||||
"{} exceeds the {} byte limit",
|
||||
path.display(),
|
||||
limit
|
||||
)));
|
||||
}
|
||||
Ok(Some(bytes))
|
||||
}
|
||||
|
||||
fn atomic_write(path: &Path, original: Option<&[u8]>, contents: &[u8]) -> io::Result<()> {
|
||||
if contents.len() as u64 > MAX_STATE_BYTES {
|
||||
return Err(invalid("Watch state exceeds the 8 MiB limit"));
|
||||
}
|
||||
let parent = path.parent().unwrap_or(Path::new("."));
|
||||
fs::create_dir_all(parent)?;
|
||||
let (temporary, mut file) = loop {
|
||||
let path = parent.join(format!(
|
||||
".watch-{}-{}.tmp",
|
||||
std::process::id(),
|
||||
NEXT_NAME.fetch_add(1, Ordering::Relaxed)
|
||||
));
|
||||
let mut options = OpenOptions::new();
|
||||
options.write(true).create_new(true);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
options.mode(0o600);
|
||||
}
|
||||
match options.open(&path) {
|
||||
Ok(file) => break (path, file),
|
||||
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
};
|
||||
let result = (|| {
|
||||
file.write_all(contents)?;
|
||||
file.sync_all()?;
|
||||
if read_bytes(path, MAX_STATE_BYTES)?.as_deref() != original {
|
||||
return Err(io::Error::other(format!(
|
||||
"{} changed on disk; restart Tubular before retrying (no changes written)",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
if original.is_none() {
|
||||
fs::hard_link(&temporary, path)?;
|
||||
} else {
|
||||
fs::rename(&temporary, path)?;
|
||||
}
|
||||
Ok(())
|
||||
})();
|
||||
drop(file);
|
||||
let _ = fs::remove_file(temporary);
|
||||
result
|
||||
}
|
||||
|
||||
fn create_private_directory(path: &Path) -> io::Result<()> {
|
||||
let mut builder = fs::DirBuilder::new();
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::DirBuilderExt;
|
||||
builder.mode(0o700);
|
||||
}
|
||||
builder.create(path)
|
||||
}
|
||||
|
||||
fn private_directory(path: &Path, create: bool) -> io::Result<()> {
|
||||
if create {
|
||||
match create_private_directory(path) {
|
||||
Ok(()) => {}
|
||||
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
let metadata = fs::symlink_metadata(path)?;
|
||||
if !metadata.is_dir() {
|
||||
return Err(invalid(format!(
|
||||
"{} must be a plain private directory, not a symlink",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
// SAFETY: geteuid has no preconditions.
|
||||
if metadata.uid() != unsafe { libc::geteuid() } || metadata.mode() & 0o077 != 0 {
|
||||
return Err(invalid(format!(
|
||||
"{} must be owned by you with permissions 0700",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn register_identity(progress_file: &Path, id: &str) -> io::Result<()> {
|
||||
if !valid_video_id(id)
|
||||
|| progress_file
|
||||
.file_name()
|
||||
.is_none_or(|name| name != "progress.jsonl")
|
||||
{
|
||||
return Err(invalid(
|
||||
"Invalid playback reporting identity or destination",
|
||||
));
|
||||
}
|
||||
let directory = progress_file
|
||||
.parent()
|
||||
.ok_or_else(|| invalid("Missing private session directory"))?;
|
||||
private_directory(directory, false)?;
|
||||
private_directory(
|
||||
directory
|
||||
.parent()
|
||||
.ok_or_else(|| invalid("Missing playback-sessions directory"))?,
|
||||
false,
|
||||
)?;
|
||||
match fs::symlink_metadata(progress_file) {
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
|
||||
Ok(_) => {
|
||||
return Err(invalid(
|
||||
"Playback report destination must not already exist",
|
||||
));
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
atomic_write(&directory.join("video-id"), None, id.as_bytes())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ProgressRecord {
|
||||
version: u32,
|
||||
event: String,
|
||||
media_url: String,
|
||||
position_seconds: f64,
|
||||
duration_seconds: Option<f64>,
|
||||
completion_eligible: bool,
|
||||
}
|
||||
|
||||
struct ProgressStream {
|
||||
path: PathBuf,
|
||||
id: Option<String>,
|
||||
file: Option<File>,
|
||||
partial: Vec<u8>,
|
||||
discarding: bool,
|
||||
failed: bool,
|
||||
line: u64,
|
||||
}
|
||||
|
||||
impl ProgressStream {
|
||||
fn new(path: PathBuf) -> Self {
|
||||
Self {
|
||||
path,
|
||||
id: None,
|
||||
file: None,
|
||||
partial: Vec::new(),
|
||||
discarding: false,
|
||||
failed: false,
|
||||
line: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn poll(
|
||||
&mut self,
|
||||
pending: &mut BTreeSet<String>,
|
||||
watched: &BTreeSet<String>,
|
||||
) -> io::Result<()> {
|
||||
if self.failed {
|
||||
return Ok(());
|
||||
}
|
||||
match self.read(pending, watched) {
|
||||
Ok(error) => match error {
|
||||
Some(e) => Err(e),
|
||||
None => Ok(()),
|
||||
},
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
|
||||
Err(e) => {
|
||||
self.failed = true;
|
||||
Err(invalid(format!("Cannot read {}: {e}", self.path.display())))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn read(
|
||||
&mut self,
|
||||
pending: &mut BTreeSet<String>,
|
||||
watched: &BTreeSet<String>,
|
||||
) -> io::Result<Option<io::Error>> {
|
||||
if self.file.is_none() {
|
||||
private_directory(self.path.parent().unwrap(), false)?;
|
||||
let marker = self.path.with_file_name("video-id");
|
||||
let mut bytes = Vec::new();
|
||||
regular_file(&marker, true)?
|
||||
.take(65)
|
||||
.read_to_end(&mut bytes)?;
|
||||
let id = std::str::from_utf8(&bytes)
|
||||
.ok()
|
||||
.filter(|id| valid_video_id(id))
|
||||
.ok_or_else(|| invalid("Invalid video-id session marker"))?;
|
||||
self.id = Some(id.into());
|
||||
self.file = Some(regular_file(&self.path, true)?);
|
||||
}
|
||||
let mut buffer = [0; STREAM_BYTES];
|
||||
let size = self.file.as_mut().unwrap().read(&mut buffer)?;
|
||||
let mut error = None;
|
||||
for &byte in &buffer[..size] {
|
||||
if byte == b'\n' {
|
||||
self.line += 1;
|
||||
if !self.discarding
|
||||
&& let Err(e) = self.record(pending, watched)
|
||||
&& error.is_none()
|
||||
{
|
||||
error = Some(e);
|
||||
}
|
||||
self.partial.clear();
|
||||
self.discarding = false;
|
||||
} else if !self.discarding {
|
||||
if self.partial.len() == MAX_LINE {
|
||||
self.partial.clear();
|
||||
self.discarding = true;
|
||||
if error.is_none() {
|
||||
error = Some(self.record_error("line exceeds 16 KiB"));
|
||||
}
|
||||
} else {
|
||||
self.partial.push(byte);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(error)
|
||||
}
|
||||
|
||||
fn record_error(&self, detail: impl std::fmt::Display) -> io::Error {
|
||||
invalid(format!(
|
||||
"Invalid playback report {} line {}: {detail}; inspect this record (other sessions continue)",
|
||||
self.path.display(),
|
||||
self.line.max(1)
|
||||
))
|
||||
}
|
||||
|
||||
fn record(&self, pending: &mut BTreeSet<String>, watched: &BTreeSet<String>) -> io::Result<()> {
|
||||
let record: ProgressRecord =
|
||||
serde_json::from_slice(&self.partial).map_err(|e| self.record_error(e))?;
|
||||
if record.version != 1 || !matches!(record.event.as_str(), "progress" | "ended" | "closed")
|
||||
{
|
||||
return Err(
|
||||
self.record_error("expected version 1 and progress, ended, or closed event")
|
||||
);
|
||||
}
|
||||
let id = record
|
||||
.media_url
|
||||
.strip_prefix("https://www.youtube.com/watch?v=")
|
||||
.filter(|id| valid_video_id(id))
|
||||
.ok_or_else(|| self.record_error("expected canonical YouTube video URL"))?;
|
||||
if !record.position_seconds.is_finite()
|
||||
|| record.position_seconds < 0.0
|
||||
|| record
|
||||
.duration_seconds
|
||||
.is_some_and(|d| !d.is_finite() || d <= 0.0)
|
||||
{
|
||||
return Err(self.record_error(
|
||||
"position must be finite/nonnegative and duration positive/finite or null",
|
||||
));
|
||||
}
|
||||
if self.id.as_deref() == Some(id)
|
||||
&& record.completion_eligible
|
||||
&& record
|
||||
.duration_seconds
|
||||
.is_some_and(|d| record.position_seconds / d >= 0.95)
|
||||
&& !watched.contains(id)
|
||||
&& !pending.contains(id)
|
||||
{
|
||||
pending.insert(id.into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
struct TestDirectory(PathBuf);
|
||||
impl TestDirectory {
|
||||
fn new() -> Self {
|
||||
loop {
|
||||
let path = std::env::temp_dir().join(format!(
|
||||
"tubular-watch-test-{}-{}",
|
||||
std::process::id(),
|
||||
NEXT_NAME.fetch_add(1, Ordering::Relaxed)
|
||||
));
|
||||
match create_private_directory(&path) {
|
||||
Ok(()) => return Self(path),
|
||||
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
|
||||
Err(e) => panic!("{e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn config(&self) -> PathBuf {
|
||||
self.0.join("config.toml")
|
||||
}
|
||||
}
|
||||
impl Drop for TestDirectory {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
fn video() -> Video {
|
||||
Video {
|
||||
id: "jNQXAC9IVRw".into(),
|
||||
title: "Me at the zoo".into(),
|
||||
channel: Some("jawed".into()),
|
||||
channel_id: Some("UC4QobU6STFB0P71PMvOGN5A".into()),
|
||||
uploader_id: Some("@jawed".into()),
|
||||
timestamp: Some(1_114_300_000),
|
||||
publication_date: Some("2005-04-23".into()),
|
||||
}
|
||||
}
|
||||
|
||||
fn append(path: &Path, bytes: &[u8]) {
|
||||
let mut options = OpenOptions::new();
|
||||
options.write(true).append(true).create(true);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
options.mode(0o600);
|
||||
}
|
||||
options.open(path).unwrap().write_all(bytes).unwrap();
|
||||
}
|
||||
|
||||
fn record(id: &str, event: &str, eligible: bool, position: f64) -> String {
|
||||
format!(
|
||||
"{{\"version\":1,\"event\":\"{event}\",\"media_url\":\"https://www.youtube.com/watch?v={id}\",\"position_seconds\":{position},\"duration_seconds\":100,\"completion_eligible\":{eligible}}}\n"
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn collections_are_independent_and_preserve_metadata_and_watched_on_restart() {
|
||||
let directory = TestDirectory::new();
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
let video = video();
|
||||
for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
|
||||
state.save(kind, &video).unwrap();
|
||||
state.save(kind, &video).unwrap();
|
||||
assert_eq!(state.videos(kind).len(), 1);
|
||||
assert!(state.contains(kind, &video.id));
|
||||
}
|
||||
let report = state.progress_file().unwrap();
|
||||
register_identity(&report, &video.id).unwrap();
|
||||
append(
|
||||
&report,
|
||||
record(&video.id, "progress", true, 95.0).as_bytes(),
|
||||
);
|
||||
assert!(state.poll_progress().unwrap());
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
let mut restarted = State::open(&directory.config()).unwrap();
|
||||
for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
|
||||
let stored = &restarted.videos(kind)[0];
|
||||
assert_eq!(stored.id, video.id);
|
||||
assert_eq!(stored.title, video.title);
|
||||
assert_eq!(stored.channel, video.channel);
|
||||
assert_eq!(stored.channel_id, video.channel_id);
|
||||
assert_eq!(stored.uploader_id, video.uploader_id);
|
||||
assert_eq!(stored.timestamp, video.timestamp);
|
||||
assert_eq!(stored.publication_date, video.publication_date);
|
||||
}
|
||||
assert!(restarted.is_watched(&video.id));
|
||||
restarted
|
||||
.remove(CollectionKind::WatchLater, std::slice::from_ref(&video.id))
|
||||
.unwrap();
|
||||
restarted
|
||||
.remove(CollectionKind::WatchLater, std::slice::from_ref(&video.id))
|
||||
.unwrap();
|
||||
let mut restarted = State::open(&directory.config()).unwrap();
|
||||
assert!(restarted.videos(CollectionKind::WatchLater).is_empty());
|
||||
assert!(restarted.contains(CollectionKind::Saved, &video.id));
|
||||
assert!(restarted.is_watched(&video.id));
|
||||
restarted.save(CollectionKind::WatchLater, &video).unwrap();
|
||||
restarted
|
||||
.remove(CollectionKind::Saved, std::slice::from_ref(&video.id))
|
||||
.unwrap();
|
||||
let mut restarted = State::open(&directory.config()).unwrap();
|
||||
assert!(restarted.videos(CollectionKind::Saved).is_empty());
|
||||
assert!(restarted.contains(CollectionKind::WatchLater, &video.id));
|
||||
assert!(restarted.is_watched(&video.id));
|
||||
assert!(!restarted.poll_progress().unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_and_externally_changed_state_is_never_overwritten() {
|
||||
let directory = TestDirectory::new();
|
||||
let path = directory.0.join("watch-later.toml");
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
fs::write(&path, "not = [valid").unwrap();
|
||||
assert!(state.save(CollectionKind::WatchLater, &video()).is_err());
|
||||
assert!(state.videos(CollectionKind::WatchLater).is_empty());
|
||||
assert_eq!(fs::read_to_string(&path).unwrap(), "not = [valid");
|
||||
assert!(State::open(&directory.config()).is_err());
|
||||
fs::remove_file(&path).unwrap();
|
||||
let saved_path = directory.0.join("saved-videos.toml");
|
||||
fs::write(&saved_path, "not = [valid").unwrap();
|
||||
assert!(state.save(CollectionKind::Saved, &video()).is_err());
|
||||
assert!(state.videos(CollectionKind::Saved).is_empty());
|
||||
assert_eq!(fs::read_to_string(&saved_path).unwrap(), "not = [valid");
|
||||
assert!(State::open(&directory.config()).is_err());
|
||||
fs::remove_file(&saved_path).unwrap();
|
||||
fs::write(&saved_path, "version = 1\n[[videos]]\nid = '--unsafe'\n").unwrap();
|
||||
assert!(State::open(&directory.config()).is_err());
|
||||
assert_eq!(
|
||||
fs::read_to_string(&saved_path).unwrap(),
|
||||
"version = 1\n[[videos]]\nid = '--unsafe'\n"
|
||||
);
|
||||
fs::remove_file(&saved_path).unwrap();
|
||||
fs::write(
|
||||
directory.0.join("watch-state.toml"),
|
||||
"version = 1\nwatched = [\"--unsafe\"]\n",
|
||||
)
|
||||
.unwrap();
|
||||
assert!(State::open(&directory.config()).is_err());
|
||||
assert_eq!(
|
||||
fs::read_to_string(directory.0.join("watch-state.toml")).unwrap(),
|
||||
"version = 1\nwatched = [\"--unsafe\"]\n"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn watched_write_failure_keeps_memory_unchanged_and_retries_consumed_reports() {
|
||||
let directory = TestDirectory::new();
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
let report = state.progress_file().unwrap();
|
||||
register_identity(&report, &video().id).unwrap();
|
||||
append(
|
||||
&report,
|
||||
record(&video().id, "progress", true, 95.0).as_bytes(),
|
||||
);
|
||||
let path = directory.0.join("watch-state.toml");
|
||||
fs::write(&path, "malformed state").unwrap();
|
||||
assert!(state.poll_progress().is_err());
|
||||
assert!(!state.is_watched(&video().id));
|
||||
assert_eq!(fs::read_to_string(&path).unwrap(), "malformed state");
|
||||
fs::remove_file(&path).unwrap();
|
||||
assert!(state.poll_progress().unwrap());
|
||||
assert!(state.is_watched(&video().id));
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn partial_reports_survive_restart_and_bad_siblings_do_not_block_completion() {
|
||||
let directory = TestDirectory::new();
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
let report = state.progress_file().unwrap();
|
||||
register_identity(&report, &video().id).unwrap();
|
||||
let line = record(&video().id, "ended", true, 95.0);
|
||||
append(&report, &line.as_bytes()[..line.len() - 1]);
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
drop(state);
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
let bad = state.progress_file().unwrap();
|
||||
register_identity(&bad, "rQ_J9WH6CGk").unwrap();
|
||||
append(&bad, b"not json\n");
|
||||
append(&report, b"\n");
|
||||
assert!(state.poll_progress().is_err());
|
||||
assert!(state.is_watched(&video().id));
|
||||
assert!(state.poll_progress().unwrap());
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wrong_identity_ineligible_positions_and_untrusted_reports_do_not_mark_watched() {
|
||||
let directory = TestDirectory::new();
|
||||
let mut state = State::open(&directory.config()).unwrap();
|
||||
let report = state.progress_file().unwrap();
|
||||
register_identity(&report, &video().id).unwrap();
|
||||
for line in [
|
||||
record("rQ_J9WH6CGk", "progress", true, 100.0),
|
||||
record(&video().id, "progress", false, 100.0),
|
||||
record(&video().id, "closed", true, 94.9),
|
||||
] {
|
||||
append(&report, line.as_bytes());
|
||||
}
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
assert!(!state.is_watched(&video().id));
|
||||
assert!(!state.is_watched("rQ_J9WH6CGk"));
|
||||
for line in [
|
||||
record(&video().id, "unknown", true, 100.0),
|
||||
record(&video().id, "progress", true, -1.0),
|
||||
record(&video().id, "progress", true, 100.0).replace("\"version\":1", "\"version\":2"),
|
||||
record(&video().id, "progress", true, 100.0)
|
||||
.replace("\"duration_seconds\":100", "\"duration_seconds\":0"),
|
||||
] {
|
||||
append(&report, line.as_bytes());
|
||||
assert!(state.poll_progress().is_err());
|
||||
assert!(!state.is_watched(&video().id));
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
}
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let symlink = state.progress_file().unwrap();
|
||||
register_identity(&symlink, &video().id).unwrap();
|
||||
let external = directory.0.join("external.jsonl");
|
||||
append(
|
||||
&external,
|
||||
record(&video().id, "progress", true, 100.0).as_bytes(),
|
||||
);
|
||||
std::os::unix::fs::symlink(&external, &symlink).unwrap();
|
||||
assert!(state.poll_progress().is_err());
|
||||
assert!(!state.is_watched(&video().id));
|
||||
assert!(!state.poll_progress().unwrap());
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user