feat(reader): a Takeout video the catalog learns later is marked watched then; the import's status

A watched video the catalog did not know at import time waits in `takeout_items`. The Shorts job now
settles those rows right after its probe (the only place a video's Short-ness becomes known), every
user's, a commit each: `reader.takeout.apply_waiting`, its count in the job's result.

`GET /api/takeout` is what the settings page reads: the user's last import run (`jobruns.last_run`)
and the kept videos by kind and outcome, "waiting" for one not known yet.

backend/app/jobruns.py

@@ -119,6 +119,16 @@ def recent_runs(db: Session, jobs: list[str], per_job: int) -> dict[str, list[Jo
     return out
 
 
+def last_run(db: Session, job: str, target: str) -> JobRun | None:
+    """The newest run of an ad-hoc `job` on `target` (a user's own import), or None."""
+    return db.execute(
+        select(JobRun)
+        .where(JobRun.job == job, JobRun.target == target)
+        .order_by(JobRun.started_at.desc(), JobRun.id.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+
+
 def gc(db: Session, retention_days: int) -> int:
     """Delete finished runs older than `retention_days` (0 = keep forever). A run still going is
     never reaped. The caller commits."""

backend/app/reader/queries.py

@@ -422,3 +422,30 @@ def takeout_outcomes(db: Session, user: User) -> dict[tuple[str, str | None], in
         .group_by(TakeoutItem.kind, TakeoutItem.outcome)
     ).all()
     return {(kind, outcome): n for kind, outcome, n in rows}
+
+
+def takeout_summary(db: Session, user: User) -> dict[str, dict[str, int]]:
+    """The user's kept Takeout videos by kind and outcome, for their settings page: `{"watched":
+    {"watched": n, "short": n, "waiting": n, …}, "watch_later": {…}}` — the import's outcome words,
+    "waiting" for a video not known yet."""
+    out: dict[str, dict[str, int]] = {kind.value: {} for kind in TakeoutKind}
+    for (kind, outcome), n in takeout_outcomes(db, user).items():
+        out[kind][outcome or "waiting"] = n
+    return out
+
+
+def takeout_users_ready(db: Session) -> list[int]:
+    """Every user with a waiting "watched" row whose video is now known and probed for Shorts: the
+    rows a later run of the Shorts probe made ready."""
+    return list(
+        db.execute(
+            select(TakeoutItem.user_id)
+            .join(Video, Video.id == TakeoutItem.video_id)
+            .where(
+                TakeoutItem.kind == TakeoutKind.WATCHED,
+                TakeoutItem.outcome.is_(None),
+                Video.shorts_probed,
+            )
+            .distinct()
+        ).scalars()
+    )

backend/app/reader/takeout.py

@@ -25,6 +25,8 @@ from contextlib import contextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
 
+from sqlalchemy.orm import Session
+
 from app import quota
 from app.bodylimit import TAKEOUT_MAX_BYTES
 from app.db import SessionLocal
@@ -174,6 +176,11 @@ def _one_import(user_id: int) -> Iterator[None]:
         yield
 
 
+def run_target(user_id: int) -> str:
+    """The import run's target, `job_runs.target`: one user's import."""
+    return f"user:{user_id}"
+
+
 def run_import(user_id: int) -> dict:
     """The import job: apply what the user's kept Takeout rows can say now, as one recorded run.
     The watched part commits before the Watch later part asks YouTube for the videos the catalog
@@ -198,3 +205,20 @@ def run_import(user_id: int) -> dict:
         if outcomes.get((TakeoutKind.WATCH_LATER, None)):
             result["watch_later_waiting"] = outcomes[(TakeoutKind.WATCH_LATER, None)]
         return result
+
+
+def apply_waiting(db: Session) -> dict:
+    """Mark watched the waiting Takeout rows whose video the catalog has come to know and the Shorts
+    probe has cleared since the import — every user's, one commit each. Run after each Shorts probe
+    (the scheduler's `shorts` job), the only place a video's Short-ness becomes known. Returns its
+    counts, or an empty dict when nothing was waiting on a probed video."""
+    users = queries.takeout_users_ready(db)
+    marked = 0
+    for user_id in users:
+        user = db.get(User, user_id)
+        if user is None:
+            continue
+        with _one_import(user_id):
+            marked += service.apply_takeout_watched(db, user)["watched"]
+            db.commit()
+    return {"takeout_watched": marked} if users else {}

backend/app/routes/feed.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.exc import StaleDataError
 
-from app import quota, scheduler
+from app import jobruns, quota, scheduler
 from app.auth import current_user
 from app.db import get_db, release_read_connection
 from app.mailtext import lang_for_user
@@ -334,7 +334,7 @@ def upload_takeout(
         scheduler.TAKEOUT_IMPORT,
         partial(takeout.run_import, user.id),
         actor_id=user.id,
-        target=f"user:{user.id}",
+        target=takeout.run_target(user.id),
         notify=True,
     )
     return {
@@ -346,3 +346,23 @@ def upload_takeout(
         "other": parsed.other,
         "malformed": parsed.malformed,
     }
+
+
+@router.get("/takeout")
+def takeout_status(user: User = Depends(current_user), db: Session = Depends(get_db)):
+    """What the user's Takeout import has done: its last run (none before the first upload), and
+    every video kept from the export, by kind and outcome — a video the catalog did not know yet
+    counts as "waiting" until it does, and is marked then."""
+    run = jobruns.last_run(db, scheduler.TAKEOUT_IMPORT, takeout.run_target(user.id))
+    return {
+        "run": None
+        if run is None
+        else {
+            "status": run.status,
+            "started_at": iso(run.started_at),
+            "finished_at": iso(run.finished_at),
+            "result": run.result,
+            "failure": run.failure,
+        },
+        **queries.takeout_summary(db, user),
+    }

backend/app/scheduler.py

@@ -22,6 +22,7 @@ from app.plex.tmdb import run_people_enrich
 from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
 from app.notifications import create_notification
 from app.runstatus import RunStatus, RunTrigger
+from app.reader import takeout
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
 from app.sync.explore import purge_ephemeral
@@ -403,7 +404,12 @@ def _autotag_job() -> None:
 
 
 def _shorts_job() -> None:
-    _job("shorts", run_shorts)
+    def work(db):
+        # A video waiting in someone's Takeout import becomes markable the moment the probe has
+        # said it is not a Short, so the import's waiting rows are settled right after it.
+        return run_shorts(db) | takeout.apply_waiting(db)
+
+    _job("shorts", work)
 
 
 def _subscriptions_job() -> None:

backend/tests/test_takeout.py

@@ -8,9 +8,10 @@ from datetime import datetime, timedelta, timezone
 import pytest
 from sqlalchemy import select
 
-from app import scheduler
+from app import jobruns, scheduler
 from app.catalog.models import Channel, Video
-from app.models import Notification
+from app.models import JobRun, Notification
+from app.runstatus import RunStatus, RunTrigger
 from app.reader import service, takeout
 from app.reader.models import (
     Playlist,
@@ -360,3 +361,83 @@ def test_an_ad_hoc_run_that_asks_for_it_tells_its_user_how_it_went(db, make_user
     assert [(n.data["job_id"], n.data["result"]) for n in notices] == [
         ("takeout_import", {"watched": 3})
     ]
+
+
+# --- the videos that wait ----------------------------------------------------------------------
+
+
+def test_waiting_videos_are_marked_once_the_shorts_probe_has_cleared_them(db, make_user):
+    a, b = make_user("a@example.com"), make_user("b@example.com")
+    _video(db, "later000000", probed=False)
+    _video(db, "notyet00000", probed=False)
+    db.commit()
+    service.stage_takeout(
+        db, a, TakeoutKind.WATCHED, {"later000000": (T0, None), "notyet00000": (T0, None)}
+    )
+    service.stage_takeout(db, b, TakeoutKind.WATCHED, {"later000000": (T0, None)})
+    db.commit()
+    assert takeout.apply_waiting(db) == {}
+
+    # The probe clears one of them; the other is still unprobed.
+    db.get(Video, "later000000").shorts_probed = True
+    db.commit()
+    assert takeout.apply_waiting(db) == {"takeout_watched": 2}
+
+    assert _states(db, a)["later000000"].watched_at == T0
+    assert _states(db, b)["later000000"].status == "watched"
+    assert _rows(db, a)["notyet00000"].outcome is None
+    assert takeout.apply_waiting(db) == {}
+
+
+def test_the_shorts_job_settles_the_waiting_rows_its_probe_made_ready(db, make_user):
+    user = make_user("job-shorts@example.com")
+    _video(db, "ready000000")  # probed already: the job's probe has nothing left to do
+    db.commit()
+    service.stage_takeout(db, user, TakeoutKind.WATCHED, {"ready000000": (T0, None)})
+    db.commit()
+
+    scheduler._shorts_job()
+
+    db.expire_all()
+    run = db.execute(select(JobRun).where(JobRun.job == "shorts")).scalar_one()
+    assert run.result == {"probed": 0, "shorts": 0, "takeout_watched": 1}
+    assert _states(db, user)["ready000000"].status == "watched"
+
+
+# --- what the settings page reads --------------------------------------------------------------
+
+
+def test_the_status_is_the_users_own_last_run_and_counts(db, make_user, as_user):
+    user, other = make_user("status@example.com"), make_user("other@example.com")
+    assert as_user(user).get("/api/takeout").json() == {
+        "run": None,
+        "watched": {},
+        "watch_later": {},
+    }
+    _video(db, "plain000000")
+    _video(db, "short000000", short=True)
+    db.commit()
+    service.stage_takeout(
+        db,
+        user,
+        TakeoutKind.WATCHED,
+        {v: (T0, None) for v in ("plain000000", "short000000", "unknown0000")},
+    )
+    service.stage_takeout(db, user, TakeoutKind.WATCH_LATER, {"plain000000": (T0, None)})
+    service.stage_takeout(db, other, TakeoutKind.WATCHED, {"plain000000": (T0, None)})
+    service.apply_takeout_watched(db, user)
+    db.commit()
+    older = jobruns.start("takeout_import", RunTrigger.AD_HOC, user.id, f"user:{user.id}")
+    jobruns.finish(older, RunStatus.OK, summary=None, result={"watched": 0})
+    newest = jobruns.start("takeout_import", RunTrigger.AD_HOC, user.id, f"user:{user.id}")
+    jobruns.finish(newest, RunStatus.OK, summary=None, result={"watched": 1})
+    theirs = jobruns.start("takeout_import", RunTrigger.AD_HOC, other.id, f"user:{other.id}")
+    jobruns.finish(theirs, RunStatus.OK, summary=None, result={"watched": 99})
+
+    body = as_user(user).get("/api/takeout").json()
+
+    assert body["run"]["status"] == "ok"
+    assert body["run"]["result"] == {"watched": 1}
+    assert body["run"]["finished_at"] is not None
+    assert body["watched"] == {"watched": 1, "short": 1, "waiting": 1}
+    assert body["watch_later"] == {"waiting": 1}

backend/tests/test_unit_fences.py

@@ -364,6 +364,8 @@ UNIT_COMMITS = {
         # The Takeout import is a recorded run with its own session: the watched part lands before
         # the Watch later part asks YouTube (which commits the videos it fetched).
         "reader/takeout.py:run_import",
+        # Run inside the Shorts job, on its session: each user's waiting rows land on their own.
+        "reader/takeout.py:apply_waiting",
     },
     # The catalog's ingest jobs commit per page or batch, a YouTube call between each: a page
     # fetched and stored is not lost when the next call fails, and an enrich that fails rolls back