diff --git a/backend/alembic/versions/0007_deep_requested.py b/backend/alembic/versions/0007_deep_requested.py new file mode 100644 index 0000000..734f473 --- /dev/null +++ b/backend/alembic/versions/0007_deep_requested.py @@ -0,0 +1,37 @@ +"""per-user opt-in to full-history (deep) backfill + +Revision ID: 0007_deep_requested +Revises: 0006_app_state +Create Date: 2026-06-11 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = "0007_deep_requested" +down_revision: Union[str, None] = "0006_app_state" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "subscriptions", + sa.Column( + "deep_requested", + sa.Boolean(), + nullable=False, + server_default="false", + ), + ) + # Preserve today's behaviour: admins already had every channel deep-backfilling + # globally, so opt their existing subscriptions in. Other users start recent-only. + op.execute( + "UPDATE subscriptions SET deep_requested = true " + "WHERE user_id IN (SELECT id FROM users WHERE role = 'admin')" + ) + + +def downgrade() -> None: + op.drop_column("subscriptions", "deep_requested") diff --git a/backend/app/models.py b/backend/app/models.py index a3e6f68..7748605 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -112,6 +112,12 @@ class Subscription(Base): yt_subscription_id: Mapped[str | None] = mapped_column(String(128)) priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0") hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + # Per-user opt-in to full-history (deep) backfill of this channel. Default off so a + # new user's unique channels don't trigger a big shared-quota burst; the deep + # scheduler only picks up channels at least one user has requested. + deep_requested: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 32c80b4..871bbff 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -54,6 +54,7 @@ def list_channels( "stored_videos": stored.get(ch.id, 0), "priority": sub.priority, "hidden": sub.hidden, + "deep_requested": sub.deep_requested, "tag_ids": tags_by_channel.get(ch.id, []), "details_synced": ch.details_synced_at is not None, "recent_synced": ch.recent_synced_at is not None, @@ -86,8 +87,18 @@ def update_channel( sub.priority = int(payload["priority"]) if "hidden" in payload: sub.hidden = bool(payload["hidden"]) + if "deep_requested" in payload: + # Opt this channel into (or out of) full-history backfill. The deep scheduler + # picks the flag up on its next run; turning it off won't undo already-fetched + # videos, it just stops further deep paging if nobody else still wants it. + sub.deep_requested = bool(payload["deep_requested"]) db.commit() - return {"id": channel_id, "priority": sub.priority, "hidden": sub.hidden} + return { + "id": channel_id, + "priority": sub.priority, + "hidden": sub.hidden, + "deep_requested": sub.deep_requested, + } @router.post("/{channel_id}/tags") diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py index a6cb902..bc5cf3d 100644 --- a/backend/app/routes/sync.py +++ b/backend/app/routes/sync.py @@ -1,12 +1,17 @@ from fastapi import APIRouter, Depends, HTTPException -from sqlalchemy import and_, case, func, select +from sqlalchemy import and_, case, func, select, update from sqlalchemy.orm import Session from app import quota, state from app.auth import current_user from app.db import get_db from app.models import Channel, Subscription, User, Video -from app.sync.runner import run_enrich, run_recent_backfill, run_rss_poll +from app.sync.runner import ( + estimate_deep_backfill, + run_enrich, + run_recent_backfill, + run_rss_poll, +) from app.sync.subscriptions import import_subscriptions router = APIRouter(prefix="/api/sync", tags=["sync"]) @@ -113,6 +118,21 @@ def my_status( ).one() total = total or 0 deep_done = int(deep_done or 0) + + # Per-user full-history (deep) backfill: channels this user has opted in, and an ETA + # for the still-pending ones (quota-bound estimate). + deep_requested_channels = ( + db.execute( + select(Channel).join( + Subscription, + and_(sub_join, Subscription.deep_requested.is_(True)), + ) + ) + .scalars() + .all() + ) + deep_eta = estimate_deep_backfill(db, deep_requested_channels) + my_videos = db.scalar( select(func.count(Video.id)).join( Subscription, @@ -130,6 +150,9 @@ def my_status( "channels_deep_done": deep_done, "channels_recent_pending": total - recent_synced, "channels_deep_pending": total - deep_done, + "channels_deep_requested": len(deep_requested_channels), + "deep_pending_count": deep_eta["channels_pending"], + "deep_eta_seconds": deep_eta["eta_seconds"], "my_videos": my_videos or 0, "quota_used_today": quota.units_used_today(db), "quota_remaining_today": quota.remaining_today(db), @@ -137,6 +160,22 @@ def my_status( } +@router.post("/deep-all") +def deep_all( + user: User = Depends(current_user), + db: Session = Depends(get_db), + on: bool = True, +) -> dict: + """Opt every one of the user's channels into (or out of) full-history backfill.""" + result = db.execute( + update(Subscription) + .where(Subscription.user_id == user.id) + .values(deep_requested=on) + ) + db.commit() + return {"updated": result.rowcount, "deep_requested": on} + + @router.post("/pause") def pause_sync( user: User = Depends(current_user), db: Session = Depends(get_db) diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index a8f87c9..acefd9a 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -13,7 +13,7 @@ from app import quota from app.config import settings log = logging.getLogger("subfeed.sync") -from app.models import Channel, OAuthToken, User +from app.models import Channel, OAuthToken, Subscription, User from app.sync.subscriptions import import_subscriptions from app.sync.videos import ( backfill_channel_deep, @@ -125,11 +125,22 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) - user = get_service_user(db) if user is None: return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"} + # Demand-driven: only deep-backfill channels at least one user has opted into + # (Subscription.deep_requested). Recent backfill stays unconditional (cheap). + deep_requested = ( + select(Subscription.id) + .where( + Subscription.channel_id == Channel.id, + Subscription.deep_requested.is_(True), + ) + .exists() + ) channels = ( db.execute( select(Channel).where( Channel.backfill_done.is_(False), Channel.recent_synced_at.is_not(None), + deep_requested, ) ) .scalars() @@ -148,3 +159,26 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) - db.rollback() log.exception("Deep backfill failed for channel %s", channel.id) return {"channels_processed": processed, "videos_added": videos_added} + + +def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict: + """Rough ETA to finish full-history backfill of `channels`. Quota-bound: total + estimated API units / the daily budget left for backfill. Deliberately approximate — + a channel's cost ~= playlistItems pages (1 unit / 50 videos) + enrichment (likewise).""" + pending = [c for c in channels if not c.backfill_done] + + def units(c: Channel) -> int: + videos = c.video_count or 0 + pages = max(1, (videos + 49) // 50) + return 2 * pages + + total_units = sum(units(c) for c in pending) + daily_budget = max(1, settings.quota_daily_budget - settings.backfill_quota_reserve) + eta_seconds = int(total_units / daily_budget * 86_400) if pending else 0 + return { + "channels_pending": len(pending), + "videos_pending_est": sum(c.video_count or 0 for c in pending), + "units_est": total_units, + "daily_budget": daily_budget, + "eta_seconds": eta_seconds, + } diff --git a/frontend/src/components/Channels.tsx b/frontend/src/components/Channels.tsx index 4540535..e0681ed 100644 --- a/frontend/src/components/Channels.tsx +++ b/frontend/src/components/Channels.tsx @@ -5,12 +5,14 @@ import { ArrowUp, Eye, EyeOff, + History, Plus, RefreshCw, Search, X, } from "lucide-react"; import { api, type ManagedChannel, type Tag } from "../lib/api"; +import { formatEta } from "../lib/format"; import { notify } from "../lib/notifications"; import Tooltip from "./Tooltip"; @@ -35,8 +37,10 @@ export default function Channels({ const userTags = (tagsQuery.data ?? []).filter((t) => !t.system); const patch = useMutation({ - mutationFn: (v: { id: string; body: { priority?: number; hidden?: boolean } }) => - api.updateChannel(v.id, v.body), + mutationFn: (v: { + id: string; + body: { priority?: number; hidden?: boolean; deep_requested?: boolean }; + }) => api.updateChannel(v.id, v.body), onSuccess: () => qc.invalidateQueries({ queryKey: ["channels"] }), }); const attach = useMutation({ @@ -66,6 +70,18 @@ export default function Channels({ mutationFn: (id: number) => api.deleteTag(id), onSuccess: () => invalidate(), }); + const deepAll = useMutation({ + mutationFn: () => api.deepAll(true), + onSuccess: (r: { updated?: number }) => { + qc.invalidateQueries({ queryKey: ["channels"] }); + qc.invalidateQueries({ queryKey: ["my-status"] }); + notify({ + level: "success", + message: `Full history requested for ${r.updated ?? 0} channels`, + }); + }, + onError: () => notify({ level: "error", message: "Couldn't request full history" }), + }); const channels = (channelsQuery.data ?? []).filter( (c) => !q || (c.title ?? "").toLowerCase().includes(q.toLowerCase()) @@ -85,9 +101,16 @@ export default function Channels({ /> + {s.deep_pending_count > 0 && ( + + )} + + +

@@ -181,6 +217,9 @@ export default function Channels({ onView={() => onViewChannel(c.id, c.title ?? "This channel")} onPriority={(d) => patch.mutate({ id: c.id, body: { priority: c.priority + d } })} onHide={() => patch.mutate({ id: c.id, body: { hidden: !c.hidden } })} + onDeep={() => + patch.mutate({ id: c.id, body: { deep_requested: !c.deep_requested } }) + } onToggleTag={(tagId) => c.tag_ids.includes(tagId) ? detach.mutate({ id: c.id, tagId }) @@ -232,6 +271,7 @@ function ChannelRow({ onView, onPriority, onHide, + onDeep, onToggleTag, }: { c: ManagedChannel; @@ -239,6 +279,7 @@ function ChannelRow({ onView: () => void; onPriority: (delta: number) => void; onHide: () => void; + onDeep: () => void; onToggleTag: (tagId: number) => void; }) { return ( @@ -279,11 +320,29 @@ function ChannelRow({ label="recent" hint={c.recent_synced ? "Recent uploads synced." : "Recent uploads not fetched yet."} /> - + {c.backfill_done ? ( + + ) : ( + + + + )} {userTags.map((t) => { const on = c.tag_ids.includes(t.id); return ( diff --git a/frontend/src/components/SettingsPanel.tsx b/frontend/src/components/SettingsPanel.tsx index 51e01a2..eb53106 100644 --- a/frontend/src/components/SettingsPanel.tsx +++ b/frontend/src/components/SettingsPanel.tsx @@ -1,8 +1,9 @@ import { useCallback, useEffect, useState, useSyncExternalStore } from "react"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; -import { Bell, Monitor, Pause, Play, RefreshCw, User, X } from "lucide-react"; +import { Bell, History, Monitor, Pause, Play, RefreshCw, User, X } from "lucide-react"; import { SCHEMES, type Scheme, type ThemePrefs } from "../lib/theme"; import { api, type Me } from "../lib/api"; +import { formatEta } from "../lib/format"; import { configureNotifications, getNotifSettings, @@ -335,6 +336,18 @@ function Sync({ me }: { me: Me }) { mutationFn: (paused: boolean) => (paused ? api.resumeSync() : api.pauseSync()), onSuccess: () => qc.invalidateQueries({ queryKey: ["my-status"] }), }); + const deepAll = useMutation({ + mutationFn: () => api.deepAll(true), + onSuccess: (r: { updated?: number }) => { + qc.invalidateQueries({ queryKey: ["my-status"] }); + qc.invalidateQueries({ queryKey: ["channels"] }); + notify({ + level: "success", + message: `Full history requested for ${r.updated ?? 0} channels`, + }); + }, + onError: () => notify({ level: "error", message: "Couldn't request full history" }), + }); return ( <> @@ -352,10 +365,18 @@ function Sync({ me }: { me: Me }) { - {`${s.channels_deep_done}/${s.channels_total}`} + {`${s.channels_deep_done}/${s.channels_deep_requested}`} + {s.deep_pending_count > 0 && ( + + {formatEta(s.deep_eta_seconds)} + + )} {s.my_videos.toLocaleString()} @@ -382,6 +403,16 @@ function Sync({ me }: { me: Me }) { Sync subscriptions + + + {me.role === "admin" && s && ( diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index ea3426a..022a594 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -144,6 +144,9 @@ export interface MyStatus { channels_deep_done: number; channels_recent_pending: number; channels_deep_pending: number; + channels_deep_requested: number; + deep_pending_count: number; + deep_eta_seconds: number; my_videos: number; quota_used_today: number; quota_remaining_today: number; @@ -160,6 +163,7 @@ export interface ManagedChannel { stored_videos: number; priority: number; hidden: boolean; + deep_requested: boolean; tag_ids: number[]; details_synced: boolean; recent_synced: boolean; @@ -184,8 +188,12 @@ export const api = { // --- channel manager --- myStatus: (): Promise => req("/api/sync/my-status"), channels: (): Promise => req("/api/channels"), - updateChannel: (id: string, patch: { priority?: number; hidden?: boolean }) => - req(`/api/channels/${id}`, { method: "PATCH", body: JSON.stringify(patch) }), + updateChannel: ( + id: string, + patch: { priority?: number; hidden?: boolean; deep_requested?: boolean } + ) => req(`/api/channels/${id}`, { method: "PATCH", body: JSON.stringify(patch) }), + deepAll: (on = true) => + req(`/api/sync/deep-all?on=${on}`, { method: "POST" }), attachChannelTag: (id: string, tagId: number) => req(`/api/channels/${id}/tags`, { method: "POST", body: JSON.stringify({ tag_id: tagId }) }), detachChannelTag: (id: string, tagId: number) => diff --git a/frontend/src/lib/format.ts b/frontend/src/lib/format.ts index 4f7b89f..440e6db 100644 --- a/frontend/src/lib/format.ts +++ b/frontend/src/lib/format.ts @@ -34,6 +34,19 @@ export function formatDuration(sec: number | null): string { return h > 0 ? `${h}:${pad(m)}:${pad(s)}` : `${m}:${pad(s)}`; } +/** Coarse, human ETA for a remaining duration in seconds ("~3 hours", "~2 days"). */ +export function formatEta(seconds: number): string { + if (seconds <= 0) return "done"; + const hours = seconds / 3600; + if (hours < 1) return "< 1 hour"; + if (hours < 24) { + const h = Math.round(hours); + return `~${h} hour${h === 1 ? "" : "s"}`; + } + const d = Math.round(hours / 24); + return `~${d} day${d === 1 ? "" : "s"}`; +} + export function formatViews(n: number | null): string { if (n == null) return ""; if (n >= 1e9) return `${(n / 1e9).toFixed(1).replace(/\.0$/, "")}B`;