Subscriptions — NewPipe-style offline subs + chronological feed

User-curated, no YouTube account. Same posture as Watch Later: you
decide what comes back, no algorithm involved.

Sidecar SubscriptionsFeed op:
  - Takes channel_ids: Vec<String>, per_channel + limit defaults 8/60
  - Caps fan-out at 200 channels per call + each id at 64 chars to
    keep a malicious caller from hammering YouTube via the sidecar
  - Spawns one tokio::task per channel against
    rustypipe::query().channel_videos(), merges results once all
    finish, sorts by publish_date string newest-first
  - A channel that 404s / region-blocks / rustypipe-errors is silently
    dropped — one dead subscription doesn't kill the whole feed; the
    failed list is returned in channels_failed for logging

Addon side:
  - subscriptions.json under addon_data, persisted via same
    _atomic_write_json + _with_lock helpers as Watch Later (no
    repeat of the race + torn-write hazards the audit caught)
  - Two new root menu entries (visible only when subscribed):
    * 'Subscriptions Feed  (N)' — chronological merge of latest uploads
    * 'Subscriptions (channel list)' — per-channel browse
  - Context menu on every video result toggles
    'Subscribe to <channel>' / 'Unsubscribe from <channel>' based on
    current sub state
  - Context menu on each entry in the channel list has its own
    'Unsubscribe from <channel>' for direct removal
  - _unsubscribe_action does Container.Refresh only when
    Container.FolderPath contains 'action=subs' (same guard pattern
    we used for wl_remove)

Live smoke (browse-only, no playback, TV in use):
  - Subscribed to LTT (UCXuqSBlHAE6Xw-yeJA0Tunw) + MKBHD
    (UCBJycsmduvYEL83R_U4JriQ) via RunPlugin
  - subscriptions.json correctly holds both
  - action=subs shows MKBHD + Linus Tech Tips as channel folders
  - action=subs_feed returns 16 merged items: MKBHD's recent uploads
    plus LTT's
  - Root menu now includes 'Subscriptions Feed  (2)' and
    'Subscriptions (channel list)'

Addon v0.0.17.
This commit is contained in:
Sulkta 2026-05-23 12:59:16 -07:00
parent 6813b27bb5
commit 13b0926860
4 changed files with 372 additions and 2 deletions

View file

@ -76,6 +76,93 @@ pub(crate) async fn channel_videos(channel_id: &str, limit: u32) -> Result<Value
}))
}
/// Fetch a subscriptions feed: pull the most recent N videos from each
/// channel in parallel, merge + sort by publish-date newest-first, cap
/// the returned union at `limit`. A channel that fails (404, region block,
/// rustypipe error) is silently dropped — one dead subscription shouldn't
/// kill the whole feed.
pub(crate) async fn subscriptions_feed(
channel_ids: &[String],
per_channel: u32,
limit: u32,
) -> Result<Value, HandlerError> {
use rustypipe::client::RustyPipe;
use std::collections::HashMap;
let rp = std::sync::Arc::new(RustyPipe::new());
// Spawn one tokio task per channel; merge results once they all finish.
// Fan-out is capped at 200 by main.rs; per_channel capped at 50.
let tasks: Vec<_> = channel_ids
.iter()
.map(|cid| {
let rp = rp.clone();
let cid = cid.clone();
tokio::spawn(async move {
let res = rp.query().channel_videos(&cid).await;
(cid, res)
})
})
.collect();
let mut all_items: Vec<Value> = Vec::new();
let mut channel_names: HashMap<String, String> = HashMap::new();
let mut failed: Vec<String> = Vec::new();
for handle in tasks {
let (cid, res) = match handle.await {
Ok(t) => t,
Err(e) => {
tracing::warn!(error = %e, "subscriptions feed task panicked");
continue;
}
};
match res {
Ok(ch) => {
if !ch.name.is_empty() {
channel_names.insert(cid.clone(), ch.name.clone());
}
for vi in ch.content.items.iter().take(per_channel as usize) {
if let Ok(v) = serde_json::to_value(vi) {
all_items.push(v);
}
}
}
Err(e) => {
tracing::warn!(channel = %cid, error = %e, "subscriptions feed channel failed");
failed.push(cid);
}
}
}
// Sort newest-first by publish_date. rustypipe VideoItem has an optional
// `publish_date` as an RFC3339-ish string when known; fall back to
// `publish_date_txt` (relative) ordering by string is meaningless, so we
// keep channel-order as a stable secondary sort by leaving channel order
// as insertion order and using a stable sort.
all_items.sort_by(|a, b| {
let ad = a.get("publish_date").and_then(Value::as_str).unwrap_or("");
let bd = b.get("publish_date").and_then(Value::as_str).unwrap_or("");
bd.cmp(ad) // descending
});
all_items.truncate(limit as usize);
tracing::info!(
channels = channel_ids.len(),
items = all_items.len(),
failed = failed.len(),
"subscriptions_feed ok"
);
Ok(serde_json::json!({
"source": "rustypipe",
"items": all_items,
"channels_total": channel_ids.len(),
"channels_failed": failed,
"channel_names": channel_names,
}))
}
/// List a playlist's videos. Returns the same VideoItem shape as search/channel.
pub(crate) async fn playlist(playlist_id: &str, limit: u32) -> Result<Value, HandlerError> {
use rustypipe::client::RustyPipe;