from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import and_, case, func, select, update from sqlalchemy.orm import Session from app import audit, quota, state from app.audit import AuditAction from app.auth import admin_user, current_user, has_read_scope, require_human from app.db import get_db from app.models import Channel, Subscription, User, Video from app.sync.runner import ( estimate_deep_backfill, run_enrich, run_recent_backfill, run_rss_poll, ) from app.sync.subscriptions import import_subscriptions from app.scheduler import running_job_ids router = APIRouter(prefix="/api/sync", tags=["sync"]) def _user_channels(db: Session, user: User) -> list[Channel]: return ( db.execute( select(Channel) .join(Subscription, Subscription.channel_id == Channel.id) .where(Subscription.user_id == user.id) ) .scalars() .all() ) @router.post("/subscriptions") def sync_subscriptions( user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: if not has_read_scope(user): # No YouTube read grant yet — the onboarding wizard hasn't been completed. raise HTTPException( status_code=403, detail="Connect your YouTube account (read access) to import subscriptions.", ) with quota.measured(db, user.id, quota.QuotaAction.SUBSCRIPTIONS_PULL) as m: result = import_subscriptions(db, user) result["quota_used_estimate"] = m.used result["quota_remaining_today"] = m.remaining return result @router.post("/rss") def sync_rss( user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: new = run_rss_poll(db, _user_channels(db, user)) return {"new_videos": new} @router.post("/backfill") def sync_backfill( user: User = Depends(require_human), db: Session = Depends(get_db), max_channels: int = 25, ) -> dict: with quota.measured(db, user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT) as m: result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels) result["quota_used_estimate"] = m.used return result @router.post("/enrich") def sync_enrich( user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: with quota.measured(db, user.id, quota.QuotaAction.VIDEOS_ENRICH) as m: enriched = run_enrich(db) return {"enriched": enriched, "quota_used_estimate": m.used} @router.get("/status") def sync_status( user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: sub_count = db.scalar( select(func.count()) .select_from(Subscription) .where(Subscription.user_id == user.id) ) 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.get("/my-status") def my_status( user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: """Per-user sync progress: how far the channels *this* user subscribes to have synced.""" sub_join = and_( Subscription.channel_id == Channel.id, Subscription.user_id == user.id ) total, details_synced, recent_synced, deep_done = db.execute( select( func.count(Channel.id), func.count(Channel.details_synced_at), func.count(Channel.recent_synced_at), func.sum(case((Channel.backfill_done.is_(True), 1), else_=0)), ) .select_from(Channel) .join(Subscription, sub_join) ).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, and_( Subscription.channel_id == Video.channel_id, Subscription.user_id == user.id, Subscription.hidden.is_(False), ), ) ) # Whole-catalog total (every video any user's subscriptions have pulled in), so the # header can show "yours" against the shared library size. total_videos = db.scalar(select(func.count(Video.id))) return { "channels_total": total, "channels_details_synced": details_synced, "channels_recent_synced": recent_synced, "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, "total_videos": total_videos or 0, "quota_used_today": quota.units_used_today(db), "quota_remaining_today": quota.remaining_today(db), "paused": state.is_sync_paused(db), # True only while a job that advances this view (channel sync) is actually running, # so the header shows live activity rather than a perpetual spinner over pending work. "sync_active": bool({"backfill", "rss_poll"} & running_job_ids()), } @router.post("/deep-all") def deep_all( user: User = Depends(require_human), 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(admin: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: state.set_sync_paused(db, True) audit.record(db, admin.id, AuditAction.SYNC_PAUSE, summary="Background sync paused") db.commit() return {"paused": True} @router.post("/resume") def resume_sync(admin: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: state.set_sync_paused(db, False) audit.record(db, admin.id, AuditAction.SYNC_RESUME, summary="Background sync resumed") db.commit() return {"paused": False}