diff --git a/backend/alembic/versions/0006_app_state.py b/backend/alembic/versions/0006_app_state.py new file mode 100644 index 0000000..2720e2a --- /dev/null +++ b/backend/alembic/versions/0006_app_state.py @@ -0,0 +1,28 @@ +"""app_state (admin-controlled global flags) + +Revision ID: 0006_app_state +Revises: 0005_shorts_probed +Create Date: 2026-06-11 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = "0006_app_state" +down_revision: Union[str, None] = "0005_shorts_probed" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "app_state", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("sync_paused", sa.Boolean(), nullable=False, server_default="false"), + ) + op.execute("INSERT INTO app_state (id, sync_paused) VALUES (1, false)") + + +def downgrade() -> None: + op.drop_table("app_state") diff --git a/backend/app/models.py b/backend/app/models.py index 84f3557..a3e6f68 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -237,3 +237,14 @@ class ApiQuotaUsage(Base): day: Mapped[date] = mapped_column(Date, primary_key=True) units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + +class AppState(Base): + """Single-row global app state (admin-controlled).""" + + __tablename__ = "app_state" + + id: Mapped[int] = mapped_column(primary_key=True, default=1) + sync_paused: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index c0d54c3..10af4de 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -1,7 +1,7 @@ from datetime import date, datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy import and_, false, func, or_, select +from sqlalchemy import Select, and_, false, func, or_, select from sqlalchemy.orm import Session, aliased from app.auth import current_user @@ -39,30 +39,28 @@ def _serialize(row) -> dict: } -@router.get("/feed") -def get_feed( - user: User = Depends(current_user), - db: Session = Depends(get_db), - tags: list[int] = Query(default=[]), - tag_mode: str = "or", - channel_id: str | None = None, - q: str | None = None, - min_duration: int | None = None, - max_duration: int | None = None, - max_age_days: int | None = None, - published_after: date | None = None, - published_before: date | None = None, - show_normal: bool = True, - include_shorts: bool = False, - include_live: bool = False, - show: str = "unwatched", # all | unwatched | watched | saved | hidden - sort: str = "newest", - seed: int = 0, - limit: int = Query(default=60, le=200), - offset: int = 0, -) -> dict: +def _filtered_query( + db: Session, + user: User, + *, + tags: list[int], + tag_mode: str, + channel_id: str | None, + q: str | None, + min_duration: int | None, + max_duration: int | None, + max_age_days: int | None, + published_after: date | None, + published_before: date | None, + show_normal: bool, + include_shorts: bool, + include_live: bool, + show: str, +) -> tuple[Select, object]: + """Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count. + Returns the column-bearing select plus the watch-status expression for sorting.""" state = aliased(VideoState) - status_expr = func.coalesce(state.status, "new").label("status") + status_expr = func.coalesce(state.status, "new") query = ( select( @@ -78,7 +76,7 @@ def get_feed( Video.view_count, Video.is_short, Video.live_status, - status_expr, + status_expr.label("status"), ) .join(Channel, Channel.id == Video.channel_id) # Only channels this user is subscribed to (and hasn't hidden). @@ -90,26 +88,9 @@ def get_feed( Subscription.hidden.is_(False), ), ) - .outerjoin( - state, and_(state.video_id == Video.id, state.user_id == user.id) - ) + .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id)) ) - # Content-type filter: Normal (regular videos incl. past-stream VODs), Shorts and - # Live/Upcoming are independent toggles; the feed is the union of the enabled types. - # The explicit Watched/Saved/Hidden views show every type so nothing goes missing. - explicit_view = show in ("watched", "saved", "hidden") - if not explicit_view: - type_clauses = [] - if show_normal: - type_clauses.append( - and_(Video.is_short.is_(False), Video.live_status.notin_(HIDDEN_LIVE)) - ) - if include_shorts: - type_clauses.append(Video.is_short.is_(True)) - if include_live: - type_clauses.append(Video.live_status.in_(HIDDEN_LIVE)) - query = query.where(or_(*type_clauses) if type_clauses else false()) if channel_id: query = query.where(Video.channel_id == channel_id) if min_duration is not None: @@ -127,27 +108,21 @@ def get_feed( ) query = query.where(Video.published_at >= start) if published_before is not None: - # Inclusive of the end day. end = datetime.combine( published_before, datetime.min.time(), tzinfo=timezone.utc ) + timedelta(days=1) query = query.where(Video.published_at < end) if q: - # Title (and channel name) only — searching descriptions produced noisy matches. like = f"%{q}%" query = query.where(or_(Video.title.ilike(like), Channel.title.ilike(like))) if tags: - # Group selected tags by category: AND across categories (e.g. language AND - # topic narrows results), OR within a category. The any/all toggle controls - # whether multiple topic tags are OR'd (any) or AND'd (all). - cat_rows = db.execute( - select(Tag.id, Tag.category).where(Tag.id.in_(tags)) - ).all() + # AND across tag categories (e.g. language AND topic narrows), OR within a + # category; the any/all toggle controls multiple topic tags. + cat_rows = db.execute(select(Tag.id, Tag.category).where(Tag.id.in_(tags))).all() by_category: dict[str, list[int]] = {} for tag_id, category in cat_rows: by_category.setdefault(category, []).append(tag_id) - visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id) for category, ids in by_category.items(): if category == "topic" and tag_mode == "and" and len(ids) > 1: @@ -163,7 +138,21 @@ def get_feed( ) query = query.where(Video.channel_id.in_(sub)) - # Watch-state visibility. + # Content type: Normal / Shorts / Live·Upcoming as a union of enabled types. + # Explicit Watched/Saved/Hidden views show every type so nothing goes missing. + explicit_view = show in ("watched", "saved", "hidden") + if not explicit_view: + type_clauses = [] + if show_normal: + type_clauses.append( + and_(Video.is_short.is_(False), Video.live_status.notin_(HIDDEN_LIVE)) + ) + if include_shorts: + type_clauses.append(Video.is_short.is_(True)) + if include_live: + type_clauses.append(Video.live_status.in_(HIDDEN_LIVE)) + query = query.where(or_(*type_clauses) if type_clauses else false()) + if show == "unwatched": query = query.where(status_expr.notin_(("watched", "hidden"))) elif show == "watched": @@ -175,22 +164,88 @@ def get_feed( else: # all query = query.where(status_expr != "hidden") - sorts = { - "newest": Video.published_at.desc().nulls_last(), - "oldest": Video.published_at.asc().nulls_last(), - "views": Video.view_count.desc().nulls_last(), - "duration_desc": Video.duration_seconds.desc().nulls_last(), - "duration_asc": Video.duration_seconds.asc().nulls_last(), - "title": func.lower(Video.title).asc().nulls_last(), - "subscribers": Channel.subscriber_count.desc().nulls_last(), - "shuffle": func.md5(func.concat(Video.id, str(seed))), + return query, status_expr + + +# Shared query parameters for /feed and /feed/count. +def _feed_params( + tags: list[int] = Query(default=[]), + tag_mode: str = "or", + channel_id: str | None = None, + q: str | None = None, + min_duration: int | None = None, + max_duration: int | None = None, + max_age_days: int | None = None, + published_after: date | None = None, + published_before: date | None = None, + show_normal: bool = True, + include_shorts: bool = False, + include_live: bool = False, + show: str = "unwatched", +) -> dict: + return { + "tags": tags, + "tag_mode": tag_mode, + "channel_id": channel_id, + "q": q, + "min_duration": min_duration, + "max_duration": max_duration, + "max_age_days": max_age_days, + "published_after": published_after, + "published_before": published_before, + "show_normal": show_normal, + "include_shorts": include_shorts, + "include_live": include_live, + "show": show, } - query = query.order_by(sorts.get(sort, sorts["newest"])) + + +SORTS = { + "newest": Video.published_at.desc().nulls_last(), + "oldest": Video.published_at.asc().nulls_last(), + "views": Video.view_count.desc().nulls_last(), + "duration_desc": Video.duration_seconds.desc().nulls_last(), + "duration_asc": Video.duration_seconds.asc().nulls_last(), + "title": func.lower(Video.title).asc().nulls_last(), + "subscribers": Channel.subscriber_count.desc().nulls_last(), +} + + +@router.get("/feed") +def get_feed( + params: dict = Depends(_feed_params), + sort: str = "newest", + seed: int = 0, + limit: int = Query(default=60, le=200), + offset: int = 0, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + query, _status = _filtered_query(db, user, **params) + order = SORTS.get(sort) + if order is None and sort == "shuffle": + order = func.md5(func.concat(Video.id, str(seed))) + query = query.order_by(order if order is not None else SORTS["newest"]) rows = db.execute(query.offset(offset).limit(limit + 1)).all() has_more = len(rows) > limit - items = [_serialize(r) for r in rows[:limit]] - return {"items": items, "has_more": has_more, "offset": offset, "limit": limit} + return { + "items": [_serialize(r) for r in rows[:limit]], + "has_more": has_more, + "offset": offset, + "limit": limit, + } + + +@router.get("/feed/count") +def get_feed_count( + params: dict = Depends(_feed_params), + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + query, _status = _filtered_query(db, user, **params) + total = db.scalar(select(func.count()).select_from(query.subquery())) + return {"count": total or 0} @router.post("/videos/{video_id}/state") diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py index e76493a..4efd941 100644 --- a/backend/app/routes/sync.py +++ b/backend/app/routes/sync.py @@ -1,8 +1,8 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import func, select from sqlalchemy.orm import Session -from app import quota +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 @@ -79,7 +79,35 @@ def sync_status( return { "subscriptions": sub_count, "channels_total": db.scalar(select(func.count()).select_from(Channel)), + "channels_backfilling": db.scalar( + select(func.count()).select_from(Channel).where(Channel.backfill_done.is_(False)) + ), "videos_total": db.scalar(select(func.count()).select_from(Video)), + "pending_enrich": db.scalar( + select(func.count()).select_from(Video).where(Video.enriched_at.is_(None)) + ), "quota_used_today": quota.units_used_today(db), "quota_remaining_today": quota.remaining_today(db), + "paused": state.is_sync_paused(db), + "is_admin": user.role == "admin", } + + +@router.post("/pause") +def pause_sync( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + if user.role != "admin": + raise HTTPException(status_code=403, detail="Admin only") + state.set_sync_paused(db, True) + return {"paused": True} + + +@router.post("/resume") +def resume_sync( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + if user.role != "admin": + raise HTTPException(status_code=403, detail="Admin only") + state.set_sync_paused(db, False) + return {"paused": False} diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index b9cc23a..8925890 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -6,6 +6,7 @@ from apscheduler.schedulers.background import BackgroundScheduler from app.config import settings from app.db import SessionLocal +from app.state import is_sync_paused from app.sync.autotag import run_autotag_all from app.sync.runner import ( run_deep_backfill, @@ -24,6 +25,9 @@ _scheduler: BackgroundScheduler | None = None def _job(name: str, fn) -> None: db = SessionLocal() try: + if is_sync_paused(db): + logger.info("job %s skipped (sync paused)", name) + return result = fn(db) logger.info("job %s -> %s", name, result) except Exception: diff --git a/backend/app/state.py b/backend/app/state.py new file mode 100644 index 0000000..3434e48 --- /dev/null +++ b/backend/app/state.py @@ -0,0 +1,24 @@ +"""Global admin-controlled app state (e.g. pausing background sync).""" +from sqlalchemy.orm import Session + +from app.models import AppState + + +def _row(db: Session) -> AppState: + row = db.get(AppState, 1) + if row is None: + row = AppState(id=1, sync_paused=False) + db.add(row) + db.commit() + return row + + +def is_sync_paused(db: Session) -> bool: + return _row(db).sync_paused + + +def set_sync_paused(db: Session, paused: bool) -> None: + row = _row(db) + row.sync_paused = paused + db.add(row) + db.commit() diff --git a/frontend/src/components/Feed.tsx b/frontend/src/components/Feed.tsx index a4270cb..a1c8c92 100644 --- a/frontend/src/components/Feed.tsx +++ b/frontend/src/components/Feed.tsx @@ -1,5 +1,5 @@ import { useEffect, useRef, useState } from "react"; -import { useInfiniteQuery, useQueryClient } from "@tanstack/react-query"; +import { useInfiniteQuery, useQuery, useQueryClient } from "@tanstack/react-query"; import { api, type FeedFilters, type Video } from "../lib/api"; import { toast } from "../lib/toast"; import VideoCard from "./VideoCard"; @@ -42,6 +42,12 @@ export default function Feed({ useEffect(() => setOverrides({}), [filters]); + const countQuery = useQuery({ + queryKey: ["feed-count", filters], + queryFn: () => api.feedCount(filters), + staleTime: 30_000, + }); + const sentinel = useRef(null); const { hasNextPage, isFetchingNextPage, fetchNextPage } = query; useEffect(() => { @@ -87,6 +93,11 @@ export default function Feed({ return (
+
+ {countQuery.data + ? `${countQuery.data.count.toLocaleString()} video${countQuery.data.count === 1 ? "" : "s"}` + : " "} +
{view === "grid" ? (
{items.map((v) => ( diff --git a/frontend/src/components/Header.tsx b/frontend/src/components/Header.tsx index 49e6b44..2524b9f 100644 --- a/frontend/src/components/Header.tsx +++ b/frontend/src/components/Header.tsx @@ -10,6 +10,7 @@ import { } from "lucide-react"; import { SCHEMES, type Scheme, type ThemePrefs } from "../lib/theme"; import type { FeedFilters, Me } from "../lib/api"; +import SyncStatus from "./SyncStatus"; function IconBtn(props: React.ButtonHTMLAttributes) { const { className = "", ...rest } = props; @@ -91,6 +92,8 @@ export default function Header({ Subfeed
+ +
(data?.paused ? api.resumeSync() : api.pauseSync()), + onSuccess: () => qc.invalidateQueries({ queryKey: ["sync-status"] }), + }); + + if (!data) return null; + + return ( +
+ + {formatViews(data.videos_total)} videos + · + {data.paused ? ( + paused + ) : data.channels_backfilling > 0 ? ( + + + {data.channels_backfilling} syncing + + ) : ( + all synced + )} + {data.is_admin && ( + + )} +
+ ); +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 7e3cf6f..3fec3fe 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -97,14 +97,30 @@ function feedQuery(f: FeedFilters, offset: number, limit: number): string { return p.toString(); } +export interface SyncStatus { + subscriptions: number; + channels_total: number; + channels_backfilling: number; + videos_total: number; + pending_enrich: number; + quota_used_today: number; + quota_remaining_today: number; + paused: boolean; + is_admin: boolean; +} + export const api = { me: (): Promise => req("/api/me"), tags: (): Promise => req("/api/tags"), - status: (): Promise => req("/api/sync/status"), + status: (): Promise => req("/api/sync/status"), feed: (f: FeedFilters, offset: number, limit: number): Promise => req(`/api/feed?${feedQuery(f, offset, limit)}`), + feedCount: (f: FeedFilters): Promise<{ count: number }> => + req(`/api/feed/count?${feedQuery(f, 0, 0)}`), setState: (id: string, status: string) => req(`/api/videos/${id}/state`, { method: "POST", body: JSON.stringify({ status }) }), + pauseSync: () => req("/api/sync/pause", { method: "POST" }), + resumeSync: () => req("/api/sync/resume", { method: "POST" }), savePrefs: (p: Record) => req("/api/me/preferences", { method: "PUT", body: JSON.stringify(p) }), };