siftlode/backend/app/routes/sync.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, case, func, select, update
from sqlalchemy.orm import Session

from app import quota, state
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(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
    state.set_sync_paused(db, True)
    return {"paused": True}


@router.post("/resume")
def resume_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
    state.set_sync_paused(db, False)
    return {"paused": False}