peter/siftlode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

merge: M5d — demand-driven deep backfill + per-user ETA

Browse source
This commit is contained in:
npeter83 2026-06-11 23:07:15 +02:00
commit f4b8a721fb
9 changed files with 256 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

37
backend/alembic/versions/0007_deep_requested.py Normal file
View file

@ -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")

6
backend/app/models.py
View file

@ -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()

13
backend/app/routes/channels.py
View file

@ -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")

43
backend/app/routes/sync.py
View file

@ -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)

36
backend/app/sync/runner.py
View file

@ -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,
}

77
frontend/src/components/Channels.tsx
View file

@ -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({
/>
<Stat
label="Full history"
value={`${s.channels_deep_done}/${s.channels_total}`}
hint="Channels whose entire back-catalog is fetched. The rest backfill gradually."
value={`${s.channels_deep_done}/${s.channels_deep_requested}`}
hint="Channels whose entire back-catalog is fetched, out of the ones you've requested full history for."
/>
{s.deep_pending_count > 0 && (
<Stat
label="left"
value={formatEta(s.deep_eta_seconds)}
hint={`Rough estimate to finish full-history backfill of ${s.deep_pending_count} pending channel(s), based on the shared daily quota.`}
/>
)}
<Stat label="My videos" value={s.my_videos.toLocaleString()} hint="Total videos available across your channels." />
<Stat
label="Quota left"
@ -121,6 +144,19 @@ export default function Channels({
Sync subscriptions
</button>
</Tooltip>
<Tooltip
side="bottom"
hint="Request full back-catalog backfill for every channel you're subscribed to. Older videos and search become complete as the shared daily quota allows — this can take a while."
>
<button
onClick={() => deepAll.mutate()}
disabled={deepAll.isPending}
className="glass-card glass-hover flex items-center gap-2 px-3 py-2 rounded-xl text-sm disabled:opacity-50 transition"
>
<History className={`w-4 h-4 ${deepAll.isPending ? "animate-pulse" : ""}`} />
Backfill everything
</button>
</Tooltip>
</div>
<p className="text-xs text-muted mb-4 leading-relaxed">
@ -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."}
/>
<SyncBadge
ok={c.backfill_done}
label="full"
hint={c.backfill_done ? "Full history fetched." : "Only recent videos so far; full history pending."}
/>
{c.backfill_done ? (
<SyncBadge ok label="full" hint="Full history fetched." />
) : (
<Tooltip
hint={
c.deep_requested
? "Full history requested — this channel's whole back-catalog will backfill as the shared quota allows. Click to cancel the request."
: "Only recent uploads so far. Click to request this channel's full back-catalog (older videos + complete search)."
}
>
<button
onClick={onDeep}
className={`inline-flex items-center gap-1 text-[10px] px-1.5 py-0.5 rounded-full border transition ${
c.deep_requested
? "bg-accent text-accent-fg border-accent"
: "bg-card border-border text-muted hover:border-accent"
}`}
>
<History className="w-3 h-3" />
{c.deep_requested ? "full history queued" : "get full history"}
</button>
</Tooltip>
)}
{userTags.map((t) => {
const on = c.tag_ids.includes(t.id);
return (

37
frontend/src/components/SettingsPanel.tsx
View file

@ -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 }) {
</Row>
<Row
label="Full history"
hint="Channels whose entire back-catalog has been fetched. The rest backfill gradually (opt in per channel)."
hint="Channels whose entire back-catalog has been fetched, out of the ones you've requested full history for (opt in per channel on the Channels page)."
>
{`${s.channels_deep_done}/${s.channels_total}`}
{`${s.channels_deep_done}/${s.channels_deep_requested}`}
</Row>
{s.deep_pending_count > 0 && (
<Row
label="Full history ETA"
hint={`Rough estimate to finish full-history backfill of ${s.deep_pending_count} pending channel(s), based on the shared daily quota.`}
>
{formatEta(s.deep_eta_seconds)}
</Row>
)}
<Row label="My videos" hint="Total videos available to you across your subscribed channels.">
{s.my_videos.toLocaleString()}
</Row>
@ -382,6 +403,16 @@ function Sync({ me }: { me: Me }) {
Sync subscriptions
</button>
</Tooltip>
<Tooltip hint="Request full back-catalog backfill for every channel you're subscribed to. Older videos + complete search fill in as the shared daily quota allows.">
<button
onClick={() => deepAll.mutate()}
disabled={deepAll.isPending}
className="glass-card glass-hover flex items-center gap-2 px-3 py-2 rounded-xl text-sm disabled:opacity-50 transition mt-2"
>
<History className={`w-4 h-4 ${deepAll.isPending ? "animate-pulse" : ""}`} />
Backfill everything
</button>
</Tooltip>
</Section>
{me.role === "admin" && s && (

12
frontend/src/lib/api.ts
View file

@ -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<MyStatus> => req("/api/sync/my-status"),
channels: (): Promise<ManagedChannel[]> => 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) =>

13
frontend/src/lib/format.ts
View file

@ -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`;