feat: integrate Miruro playback and episode progress

This commit is contained in:
2026-10-12 22:23:15 +02:00
parent 0fe24ff302
commit 0e909001d0
9 changed files with 1676 additions and 92 deletions
+507 -28
View File
@@ -10,11 +10,14 @@ use serde::Deserialize;
use toml_edit::{Array, ArrayOfTables, DocumentMut, Table, value};
use crate::youtube::{Video, valid_video_id, validate_video};
use tubular::miruro::{Identity, SourceDocument};
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;
const MAX_SOURCE_BYTES: usize = 1024 * 1024;
const MAX_IDENTITY_BYTES: u64 = 7 + 512 + 1 + 10;
static NEXT_NAME: AtomicU64 = AtomicU64::new(0);
fn invalid(message: impl Into<String>) -> io::Error {
@@ -239,6 +242,26 @@ impl State {
Ok(path)
}
pub fn miruro_session(
&mut self,
video: &Video,
source: &SourceDocument,
) -> io::Result<(PathBuf, PathBuf)> {
source.validate()?;
if Identity::from_key(&video.id)? != source.identity {
return Err(invalid("Source identity must match the playback episode"));
}
let contents = serde_json::to_vec(source).map_err(|e| invalid(e.to_string()))?;
if contents.len() > MAX_SOURCE_BYTES {
return Err(invalid("Playback source exceeds the 1 MiB limit"));
}
let progress_file = self.progress_file(video)?;
let source_file = progress_file.with_file_name("source.json");
atomic_write(&source_file, None, &contents)?;
register_identity(&progress_file, &video.id)?;
Ok((source_file, progress_file))
}
pub fn poll_progress(&mut self) -> io::Result<bool> {
let mut error = self.discover().err();
let count = self.streams.len().min(16);
@@ -372,6 +395,23 @@ fn validate_snapshot(video: &mut Video, require_channel: bool) -> io::Result<()>
));
}
}
if video.id.starts_with("miruro:") {
Identity::from_key(&video.id)?;
if video.title.trim().is_empty() || video.title.chars().any(char::is_control) {
return Err(invalid(
"Miruro episode title must be nonempty and contain no controls",
));
}
if video.channel.as_deref() != Some("Miruro")
|| video.channel_id.is_some()
|| video.uploader_id.is_some()
{
return Err(invalid(
"Miruro episodes must not contain YouTube channel identities",
));
}
return Ok(());
}
validate_video(video).map_err(invalid)?;
video.publication_date = date;
if let Some(handle) = &video.uploader_id
@@ -448,13 +488,13 @@ fn load_watched(document: &DocumentMut) -> io::Result<BTreeSet<String>> {
return Ok(BTreeSet::new());
};
item.as_array()
.ok_or_else(|| invalid("watched must be an array of YouTube IDs"))?
.ok_or_else(|| invalid("watched must be an array of playback identities"))?
.iter()
.map(|v| {
let id = v
.as_str()
.filter(|id| valid_video_id(id))
.ok_or_else(|| invalid("Invalid watched YouTube ID"))?;
.filter(|id| valid_identity(id))
.ok_or_else(|| invalid("Invalid watched playback identity"))?;
Ok(id.to_owned())
})
.collect()
@@ -641,8 +681,12 @@ fn private_directory(path: &Path, create: bool) -> io::Result<()> {
Ok(())
}
fn valid_identity(id: &str) -> bool {
valid_video_id(id) || Identity::from_key(id).is_ok()
}
pub(crate) fn register_identity(progress_file: &Path, id: &str) -> io::Result<()> {
if !valid_video_id(id)
if !valid_identity(id)
|| progress_file
.file_name()
.is_none_or(|name| name != "progress.jsonl")
@@ -674,7 +718,14 @@ pub(crate) fn register_identity(progress_file: &Path, id: &str) -> io::Result<()
}
#[derive(Deserialize)]
struct ProgressRecord {
#[serde(untagged)]
enum ProgressRecord {
YouTube(YouTubeProgress),
Miruro(MiruroProgress),
}
#[derive(Deserialize)]
struct YouTubeProgress {
version: u32,
event: String,
media_url: String,
@@ -683,6 +734,17 @@ struct ProgressRecord {
completion_eligible: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct MiruroProgress {
version: u32,
event: String,
media_identity: Identity,
position_seconds: f64,
duration_seconds: Option<f64>,
completion_eligible: bool,
}
struct ProgressStream {
path: PathBuf,
id: Option<String>,
@@ -722,11 +784,11 @@ impl ProgressStream {
let marker = self.path.with_file_name("video-id");
let mut bytes = Vec::new();
regular_file(&marker, true)?
.take(65)
.take(MAX_IDENTITY_BYTES + 1)
.read_to_end(&mut bytes)?;
let id = std::str::from_utf8(&bytes)
.ok()
.filter(|id| valid_video_id(id))
.filter(|id| id.len() as u64 <= MAX_IDENTITY_BYTES && valid_identity(id))
.ok_or_else(|| invalid("Invalid video-id session marker"))?;
let snapshot = self.path.with_file_name("video.toml");
let video = match regular_file(&snapshot, true) {
@@ -739,7 +801,12 @@ impl ProgressStream {
}
collection.videos.pop()
}
Err(e) if e.kind() == io::ErrorKind::NotFound => None,
Err(e) if e.kind() == io::ErrorKind::NotFound && valid_video_id(id) => None,
Err(e) if e.kind() == io::ErrorKind::NotFound => {
return Err(invalid(
"Miruro sessions require their bound video metadata",
));
}
Err(e) => return Err(e),
};
let seen_path = self.path.with_file_name("watching-seen.toml");
@@ -856,22 +923,48 @@ impl ProgressStream {
) -> 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 miruro_key;
let (id, event, position, duration, eligible) = match &record {
ProgressRecord::YouTube(record) if record.version == 1 => {
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"))?;
(
id,
record.event.as_str(),
record.position_seconds,
record.duration_seconds,
record.completion_eligible,
)
}
ProgressRecord::Miruro(record) if record.version == 2 => {
record
.media_identity
.validate()
.map_err(|e| self.record_error(e))?;
miruro_key = record.media_identity.key();
(
miruro_key.as_str(),
record.event.as_str(),
record.position_seconds,
record.duration_seconds,
record.completion_eligible,
)
}
_ => {
return Err(self.record_error(
"expected version 1 with media_url or version 2 with media_identity",
));
}
};
if !matches!(event, "progress" | "ended" | "closed") {
return Err(self.record_error("expected 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)
if !position.is_finite()
|| position < 0.0
|| duration.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",
@@ -880,13 +973,11 @@ impl ProgressStream {
if self.id.as_deref() != Some(id) {
return Ok(());
}
if record.position_seconds > 0.0 && !self.seen && !watched.contains(id) {
if position > 0.0 && !self.seen && !watched.contains(id) {
self.started = true;
}
if record.completion_eligible
&& record
.duration_seconds
.is_some_and(|d| record.position_seconds / d >= 0.95)
if eligible
&& duration.is_some_and(|d| position / d >= 0.95)
&& !watched.contains(id)
&& !pending.contains(id)
{
@@ -955,6 +1046,50 @@ mod tests {
)
}
fn miruro_video(episode: u32) -> Video {
Video {
id: Identity::new("opaque_Show-1".into(), episode)
.unwrap()
.key(),
title: format!("Example show · Episode {episode}"),
channel: Some("Miruro".into()),
channel_id: None,
uploader_id: None,
timestamp: Some(1_114_300_000),
publication_date: Some("2005-04-23".into()),
}
}
fn miruro_source(video: &Video) -> SourceDocument {
SourceDocument {
version: 1,
url: "https://cdn.example/video.m3u8?token=private-token".into(),
headers: [("Referer".into(), "https://player.example/".into())].into(),
identity: Identity::from_key(&video.id).unwrap(),
title: Some(video.title.clone()),
subtitles: Vec::new(),
}
}
fn miruro_record(
video: &Video,
event: &str,
eligible: bool,
position: f64,
duration: Option<f64>,
) -> String {
serde_json::json!({
"version": 2,
"event": event,
"media_identity": Identity::from_key(&video.id).unwrap(),
"position_seconds": position,
"duration_seconds": duration,
"completion_eligible": eligible,
})
.to_string()
+ "\n"
}
#[test]
fn collections_are_independent_and_preserve_metadata_and_watched_on_restart() {
let directory = TestDirectory::new();
@@ -1565,4 +1700,348 @@ mod tests {
.unwrap();
assert!(State::open(&directory.config()).is_err());
}
#[test]
fn miruro_sessions_recover_collections_and_watching_without_persisting_sources_in_state() {
let directory = TestDirectory::new();
let video = miruro_video(1);
let mut state = State::open(&directory.config()).unwrap();
for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
state.save(kind, &video).unwrap();
}
let source = miruro_source(&video);
let (source_file, report) = state.miruro_session(&video, &source).unwrap();
let source_bytes = fs::read(&source_file).unwrap();
assert_eq!(source_bytes, serde_json::to_vec(&source).unwrap());
assert!(!report.exists());
assert_eq!(
fs::read_to_string(report.with_file_name("video-id")).unwrap(),
video.id
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
for path in [
&source_file,
&report.with_file_name("video-id"),
&report.with_file_name("video.toml"),
] {
assert_eq!(
fs::metadata(path).unwrap().permissions().mode() & 0o777,
0o600
);
}
for path in [
report.parent().unwrap(),
&directory.0.join("playback-sessions"),
] {
assert_eq!(
fs::metadata(path).unwrap().permissions().mode() & 0o777,
0o700
);
}
}
assert!(!state.poll_progress().unwrap());
assert!(state.videos(CollectionKind::Watching).is_empty());
drop(state);
// A GUI can acknowledge its final snapshot after Tubular has quit.
append(
&report,
miruro_record(&video, "closed", false, 10.0, None).as_bytes(),
);
let mut state = State::open(&directory.config()).unwrap();
assert!(state.poll_progress().unwrap());
let stored = &state.videos(CollectionKind::Watching)[0];
assert_eq!(stored.id, video.id);
assert_eq!(stored.title, video.title);
assert_eq!(stored.channel, video.channel);
assert_eq!(stored.timestamp, video.timestamp);
assert_eq!(stored.publication_date, video.publication_date);
drop(state);
let mut state = State::open(&directory.config()).unwrap();
assert!(state.contains(CollectionKind::Watching, &video.id));
assert!(!state.poll_progress().unwrap());
append(
&report,
miruro_record(&video, "ended", true, 95.0, Some(100.0)).as_bytes(),
);
assert!(state.poll_progress().unwrap());
assert!(state.is_watched(&video.id));
assert!(state.videos(CollectionKind::Watching).is_empty());
drop(state);
let mut state = State::open(&directory.config()).unwrap();
assert!(state.is_watched(&video.id));
assert!(!state.poll_progress().unwrap());
for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
let stored = &state.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, None);
assert_eq!(stored.uploader_id, None);
assert_eq!(stored.timestamp, video.timestamp);
assert_eq!(stored.publication_date, video.publication_date);
}
for path in [
directory.0.join("watch-later.toml"),
directory.0.join("saved-videos.toml"),
directory.0.join("currently-watching.toml"),
directory.0.join("watch-state.toml"),
report.with_file_name("video.toml"),
] {
let contents = fs::read_to_string(path).unwrap();
assert!(!contents.contains("private-token"));
assert!(!contents.contains("cdn.example"));
}
assert_eq!(fs::read(source_file).unwrap(), source_bytes);
assert!(report.exists());
}
#[test]
fn miruro_v2_requires_matching_episode_eligible_known_duration_and_threshold() {
for event in ["progress", "ended", "closed"] {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let video = miruro_video(1);
let (_, report) = state
.miruro_session(&video, &miruro_source(&video))
.unwrap();
append(
&report,
miruro_record(&miruro_video(2), event, true, 100.0, Some(100.0)).as_bytes(),
);
assert!(!state.poll_progress().unwrap());
assert!(state.videos(CollectionKind::Watching).is_empty());
assert!(!state.is_watched(&miruro_video(2).id));
for (eligible, position, duration) in [
(true, 0.0, Some(100.0)),
(true, 94.9, Some(100.0)),
(false, 100.0, Some(100.0)),
(true, 100.0, None),
] {
append(
&report,
miruro_record(&video, event, eligible, position, duration).as_bytes(),
);
state.poll_progress().unwrap();
assert!(!state.is_watched(&video.id));
}
assert!(state.contains(CollectionKind::Watching, &video.id));
let completion = miruro_record(&video, event, true, 95.0, Some(100.0));
append(&report, completion.as_bytes());
append(&report, completion.as_bytes());
assert!(state.poll_progress().unwrap());
assert!(state.is_watched(&video.id));
assert!(state.videos(CollectionKind::Watching).is_empty());
assert!(!state.poll_progress().unwrap());
}
}
#[test]
fn miruro_v2_rejects_invalid_wire_schemas_numbers_and_namespaces() {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let video = miruro_video(1);
let (_, report) = state
.miruro_session(&video, &miruro_source(&video))
.unwrap();
let valid = miruro_record(&video, "progress", true, 100.0, Some(100.0));
for line in [
valid.replace("\"version\":2", "\"version\":1"),
valid.replace("\"version\":2", "\"version\":3"),
valid.replace("\"provider\":\"miruro\"", "\"provider\":\"youtube\""),
valid.replace("\"episode\":1", "\"episode\":0"),
valid.replace("\"episode\":1", "\"episode\":2147483648"),
valid.replace("\"position_seconds\":100.0", "\"position_seconds\":-1"),
valid.replace("\"position_seconds\":100.0", "\"position_seconds\":1e999"),
valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":0"),
valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":-1"),
valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":1e999"),
valid.replace("\"event\":\"progress\"", "\"event\":\"other\""),
valid.replacen('{', "{\"media_url\":null,", 1),
record(&self::video().id, "ended", true, 100.0)
.replace("\"version\":1", "\"version\":2"),
] {
append(&report, line.as_bytes());
assert!(state.poll_progress().is_err(), "{line}");
assert!(state.videos(CollectionKind::Watching).is_empty());
assert!(!state.is_watched(&video.id));
assert!(!state.poll_progress().unwrap());
}
}
#[test]
fn miruro_partial_reports_and_bad_siblings_recover_after_restart() {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let video = miruro_video(1);
let (source, report) = state
.miruro_session(&video, &miruro_source(&video))
.unwrap();
let completion = miruro_record(&video, "closed", true, 95.0, Some(100.0));
append(&report, &completion.as_bytes()[..completion.len() - 1]);
assert!(!state.poll_progress().unwrap());
drop(state);
let mut state = State::open(&directory.config()).unwrap();
assert!(!state.poll_progress().unwrap());
let other = miruro_video(2);
let (_, bad) = state
.miruro_session(&other, &miruro_source(&other))
.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.is_watched(&other.id));
assert!(state.poll_progress().unwrap());
assert!(!state.poll_progress().unwrap());
drop(state);
assert!(source.exists());
assert!(report.exists());
let state = State::open(&directory.config()).unwrap();
assert!(state.is_watched(&video.id));
}
#[test]
fn miruro_state_rejects_catalog_rows_bad_titles_and_youtube_metadata() {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
for field in [
"series",
"noncanonical",
"empty-title",
"control-title",
"channel",
"channel-id",
"uploader",
] {
let mut video = miruro_video(1);
match field {
"series" => video.id = "miruro-series:opaque_Show-1".into(),
"noncanonical" => video.id = "miruro:opaque_Show-1:01".into(),
"empty-title" => video.title = " ".into(),
"control-title" => video.title.push('\n'),
"channel" => video.channel = Some("YouTube".into()),
"channel-id" => video.channel_id = self::video().channel_id,
"uploader" => video.uploader_id = Some("@jawed".into()),
_ => unreachable!(),
}
for kind in [
CollectionKind::WatchLater,
CollectionKind::Saved,
CollectionKind::Watching,
] {
assert!(state.save(kind, &video).is_err(), "{field}");
}
assert!(state.progress_file(&video).is_err(), "{field}");
}
for id in [
"miruro-series:opaque_Show-1",
"miruro:opaque_Show-1:01",
"other:opaque_Show-1:1",
] {
let mut watched = DocumentMut::new();
watched["version"] = value(1);
let mut ids = Array::new();
ids.push(id);
watched["watched"] = value(ids);
assert!(load_watched(&watched).is_err());
}
}
#[test]
fn miruro_source_sessions_validate_binding_size_and_maximum_identity_marker() {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let mut video = miruro_video(i32::MAX as u32);
video.id = Identity::new("a".repeat(512), i32::MAX as u32)
.unwrap()
.key();
let mut source = miruro_source(&video);
source.identity.episode -= 1;
assert!(state.miruro_session(&video, &source).is_err());
assert!(state.streams.is_empty());
source.identity = Identity::from_key(&video.id).unwrap();
source.title = Some("x".repeat(MAX_SOURCE_BYTES));
assert!(state.miruro_session(&video, &source).is_err());
assert!(state.streams.is_empty());
source.title = None;
let (_, report) = state.miruro_session(&video, &source).unwrap();
append(
&report,
miruro_record(&video, "ended", true, 95.0, Some(100.0)).as_bytes(),
);
drop(state);
let mut state = State::open(&directory.config()).unwrap();
assert!(state.poll_progress().unwrap());
assert!(state.is_watched(&video.id));
drop(state);
let state = State::open(&directory.config()).unwrap();
assert!(state.is_watched(&video.id));
}
#[test]
fn miruro_recovery_rejects_missing_snapshots_and_overlong_identity_markers() {
for missing_snapshot in [true, false] {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let video = miruro_video(1);
let (_, report) = state
.miruro_session(&video, &miruro_source(&video))
.unwrap();
if missing_snapshot {
fs::remove_file(report.with_file_name("video.toml")).unwrap();
} else {
append(
&report.with_file_name("video-id"),
&vec![b'x'; MAX_IDENTITY_BYTES as usize],
);
}
append(
&report,
miruro_record(&video, "ended", true, 100.0, Some(100.0)).as_bytes(),
);
drop(state);
let mut state = State::open(&directory.config()).unwrap();
assert!(state.poll_progress().is_err());
assert!(!state.is_watched(&video.id));
assert!(state.videos(CollectionKind::Watching).is_empty());
}
}
#[test]
fn youtube_v1_reports_keep_canonical_urls_and_unknown_field_tolerance() {
let directory = TestDirectory::new();
let mut state = State::open(&directory.config()).unwrap();
let video = video();
let report = state.progress_file(&video).unwrap();
register_identity(&report, &video.id).unwrap();
let valid = record(&video.id, "ended", true, 95.0);
for url in [
format!("https://youtu.be/{}", video.id),
format!("https://www.youtube.com/watch?v={}&t=1", video.id),
format!("http://www.youtube.com/watch?v={}", video.id),
] {
append(
&report,
valid
.replace(
&format!("https://www.youtube.com/watch?v={}", video.id),
&url,
)
.as_bytes(),
);
assert!(state.poll_progress().is_err());
assert!(!state.is_watched(&video.id));
assert!(state.videos(CollectionKind::Watching).is_empty());
}
append(
&report,
valid
.replacen('{', "{\"legacy_extension\":true,", 1)
.as_bytes(),
);
assert!(state.poll_progress().unwrap());
assert!(state.is_watched(&video.id));
}
}