diff --git a/backend/app/bodylimit.py b/backend/app/bodylimit.py index 97b67eba..432c6e1c 100644 --- a/backend/app/bodylimit.py +++ b/backend/app/bodylimit.py @@ -72,9 +72,11 @@ DEFAULT_MAX_BODY_BYTES = 1024 * 1024 # A route belongs here ONLY IF ITS OWN CAP IS LARGER THAN THE DEFAULT. That is why # `POST /api/diagnostics/events` is absent: its `capped_body` limit is 512 KiB, already below the # default, so listing it could only widen what it accepts. -# A Google Takeout file (`reader.takeout`): a real watch history of 9500 views, the 18 months Google -# keeps by default, is 3.7 MB; four times that is still well inside one request. -TAKEOUT_MAX_BYTES = 16 * 1024 * 1024 +# A Google Takeout file (`reader.takeout`). A real watch history measured 3.7 MB for 9500 views, ~390 +# bytes a view, so 32 MiB holds ~86k views. YouTube keeps the history 36 months by default on an +# account made since mid-2020 and for good on an older one, so a heavy viewer's can pass 16 MiB. The +# file is read an entry at a time (`takeout._json_list`), so it costs its own size, not a tree of it. +TAKEOUT_MAX_BYTES = 32 * 1024 * 1024 ROUTE_MAX_BODY_BYTES: dict[tuple[str, str], int] = { # Ingesting a local file. The body IS the file now — no multipart envelope to allow for — diff --git a/backend/app/reader/queries.py b/backend/app/reader/queries.py index 88a6550a..9ac01add 100644 --- a/backend/app/reader/queries.py +++ b/backend/app/reader/queries.py @@ -7,6 +7,7 @@ length, whether a video exists); another unit asks the questions at the end of t than joining Reader's tables itself. """ +from collections.abc import Iterable from datetime import datetime from sqlalchemy import ColumnElement, and_, func, or_, select @@ -407,11 +408,41 @@ def takeout_waiting(db: Session, user: User, kind: TakeoutKind) -> list[tuple[in ) +# Ids per IN list: a list is bound one parameter an id, and a statement takes at most 65,535. +_IN_CHUNK = 5000 + + def known_video_ids(db: Session, ids: list[str]) -> set[str]: """Which of `ids` the catalog holds.""" - if not ids: - return set() - return set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars()) + known: set[str] = set() + for start in range(0, len(ids), _IN_CHUNK): + chunk = ids[start : start + _IN_CHUNK] + known.update(db.execute(select(Video.id).where(Video.id.in_(chunk))).scalars()) + return known + + +def takeout_rows_besides( + db: Session, user: User, kind: TakeoutKind, video_ids: Iterable[str] +) -> int: + """How many Takeout rows the user keeps other than `video_ids` of `kind`: what an upload of + those videos would be added to.""" + total = ( + db.scalar(select(func.count(TakeoutItem.id)).where(owned_by(TakeoutItem.user_id, user))) + or 0 + ) + ids = list(video_ids) + for start in range(0, len(ids), _IN_CHUNK): + total -= ( + db.scalar( + select(func.count(TakeoutItem.id)).where( + owned_by(TakeoutItem.user_id, user), + TakeoutItem.kind == kind, + TakeoutItem.video_id.in_(ids[start : start + _IN_CHUNK]), + ) + ) + or 0 + ) + return total def takeout_outcomes(db: Session, user: User) -> dict[tuple[str, str | None], int]: @@ -449,3 +480,14 @@ def takeout_users_ready(db: Session) -> list[int]: .distinct() ).scalars() ) + + +def takeout_users_watch_later_waiting(db: Session) -> list[int]: + """Every user with a Watch later row a run could not put in yet.""" + return list( + db.execute( + select(TakeoutItem.user_id) + .where(TakeoutItem.kind == TakeoutKind.WATCH_LATER, TakeoutItem.outcome.is_(None)) + .distinct() + ).scalars() + ) diff --git a/backend/app/reader/service.py b/backend/app/reader/service.py index 7113eaae..4d396aee 100644 --- a/backend/app/reader/service.py +++ b/backend/app/reader/service.py @@ -510,6 +510,15 @@ def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> No # --- Google Takeout -------------------------------------------------------------------------- _TAKEOUT_BATCH = 1000 +# The advisory-lock namespace of one user's Takeout rows ("TO"), so a key cannot meet another lock's. +_TAKEOUT_LOCK_NS = 0x544F + + +def lock_takeout(db: Session, user: User) -> None: + """Hold the user's Takeout rows for the rest of the transaction. An upload's staging and an + import's settling each lock many of the same rows, one statement after another and in no common + order, so without one lock over both the two can deadlock.""" + db.execute(select(func.pg_advisory_xact_lock(_TAKEOUT_LOCK_NS, user.id))) def stage_takeout( @@ -518,6 +527,7 @@ def stage_takeout( """Keep the videos of one Takeout file, `{video id: (time, channel id)}`, for the import job. A second upload of the same export, or a newer one, adds what is new: a video already kept keeps the later time, and a row the import has settled keeps its outcome.""" + lock_takeout(db, user) values = [ {"user_id": user.id, "kind": kind, "video_id": vid, "at": at, "channel_id": channel} for vid, (at, channel) in rows.items() @@ -551,6 +561,7 @@ def apply_takeout_watched(db: Session, user: User) -> dict[str, int]: past is not watching it), and a state the user set here wins: watched or hidden, or anything changed after the Takeout time. The upsert's own condition is that rule, so a state set while the import runs wins too.""" + lock_takeout(db, user) short: list[int] = [] todo: dict[str, tuple[int, datetime]] = {} for row_id, video_id, at, is_short in queries.takeout_watched_ready(db, user): @@ -590,19 +601,22 @@ def apply_takeout_watched(db: Session, user: User) -> dict[str, int]: return {"watched": len(marked), "kept": len(kept), "shorts": len(short)} -def apply_takeout_watch_later(db: Session, user: User) -> dict[str, int]: - """Put the videos of the user's YouTube Watch later list into theirs here, oldest first, after - what it already holds. Called once the catalog has fetched what it could: a video still unknown - is one YouTube no longer has.""" - waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER) - if not waiting: +def apply_takeout_watch_later( + db: Session, user: User, rows: list[tuple[int, str]], asked: set[str] +) -> dict[str, int]: + """Put the videos of the user's waiting Watch later `rows`, (row id, video id) oldest first, into + their list here, after what it already holds. `asked` is the videos YouTube was just asked for: one + of those the catalog still lacks is gone from YouTube; any other unknown one keeps waiting.""" + if not rows: return {} - known = queries.known_video_ids(db, [vid for _, vid in waiting]) + lock_takeout(db, user) + known = queries.known_video_ids(db, [vid for _, vid in rows]) pl = watch_later(db, user) done: dict[TakeoutOutcome, list[int]] = {o: [] for o in TakeoutOutcome} - for row_id, vid in waiting: + for row_id, vid in rows: if vid not in known: - done[TakeoutOutcome.MISSING].append(row_id) + if vid in asked: + done[TakeoutOutcome.MISSING].append(row_id) elif add_item(db, pl, vid, mark_dirty=False): done[TakeoutOutcome.ADDED].append(row_id) else: diff --git a/backend/app/reader/takeout.py b/backend/app/reader/takeout.py index 7a7b12b9..67231e33 100644 --- a/backend/app/reader/takeout.py +++ b/backend/app/reader/takeout.py @@ -24,6 +24,7 @@ from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime +from itertools import chain from sqlalchemy.orm import Session @@ -31,6 +32,7 @@ from app import quota from app.bodylimit import TAKEOUT_MAX_BYTES from app.db import SessionLocal from app.models import User +from app.ratelimit import RateLimiter from app.reader import queries, service from app.reader.models import TakeoutKind from app.sync.runner import get_service_user @@ -39,6 +41,12 @@ from app.youtube.client import YouTubeClient # The largest file the upload takes (its edge cap and the route's own are one number). MAX_BYTES = TAKEOUT_MAX_BYTES +# The most videos a Watch later file may hold: YouTube's own cap on one playlist. Each unknown one is +# looked up on the shared API budget, so a longer list is not one YouTube made. +WATCH_LATER_MAX = 5000 +# The most Takeout videos one user keeps: twenty times the real export measured here (~9,000 videos +# across both files), so it only stops uploads of made-up ones piling up for good. +ROWS_MAX = 200_000 _VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$") _WATCH_URL = re.compile( @@ -84,7 +92,7 @@ def _keep(parsed: Parsed, vid: str, at: datetime, channel: str | None) -> None: parsed.videos[vid] = (at, channel or (old[1] if old else None)) -def _history(data: list) -> Parsed: +def _history(data: Iterator[object]) -> Parsed: parsed = Parsed(TakeoutKind.WATCHED) for entry in data: parsed.entries += 1 @@ -117,22 +125,54 @@ def _history(data: list) -> Parsed: def _watch_later(text: str) -> Parsed: parsed = Parsed(TakeoutKind.WATCH_LATER) - rows = [row for row in csv.reader(io.StringIO(text)) if any(cell.strip() for cell in row)] - if not rows or len(rows[0]) < 2: + rows = (row for row in csv.reader(io.StringIO(text)) if any(cell.strip() for cell in row)) + first = next(rows, None) + if first is None or len(first) < 2: raise TakeoutError("not_takeout") - if not _VIDEO_ID.match(rows[0][0].strip()): - rows = rows[1:] # the header, in the account's language - for row in rows: + # The first row is the header, in the account's language, unless it is already a video. + for row in chain([first], rows) if _VIDEO_ID.match(first[0].strip()) else rows: parsed.entries += 1 - vid = row[0].strip() if row else "" + vid = row[0].strip() at = _time(row[1].strip()) if len(row) > 1 else None if not _VIDEO_ID.match(vid) or at is None: parsed.malformed += 1 continue _keep(parsed, vid, at, None) + if len(parsed.videos) > WATCH_LATER_MAX: + raise TakeoutError("too_many") return parsed +def _skip(text: str, at: int) -> int: + """The first index at or after `at` that is not JSON whitespace.""" + while at < len(text) and text[at] in " \t\n\r": + at += 1 + return at + + +def _json_list(text: str) -> Iterator[object]: + """The entries of a JSON list, one at a time: the whole tree is never built, so a file of a + million tiny entries costs its text and one entry, not a million dicts at once.""" + decoder = json.JSONDecoder() + at = _skip(text, _skip(text, 0) + 1) # past the "[" the caller saw + if text.startswith("]", at): + at += 1 + else: + while True: + entry, at = decoder.raw_decode(text, at) + yield entry + at = _skip(text, at) + if text.startswith(",", at): + at = _skip(text, at + 1) + elif text.startswith("]", at): + at += 1 + break + else: + raise ValueError("not a JSON list") + if _skip(text, at) != len(text): + raise ValueError("text after the JSON list") + + def parse(body: bytes) -> Parsed: """Read an uploaded Takeout file: the watch history (JSON) or the Watch later list (csv), told apart by their content, not their name. Raises `TakeoutError` for anything else, and for a file @@ -142,19 +182,19 @@ def parse(body: bytes) -> Parsed: except UnicodeDecodeError: raise TakeoutError("not_takeout") from None stripped = text.lstrip() - if stripped.startswith("["): - try: - data = json.loads(stripped) - except json.JSONDecodeError: - raise TakeoutError("not_takeout") from None - if not isinstance(data, list): - raise TakeoutError("not_takeout") - parsed = _history(data) - elif stripped.startswith(("{", "<")): + if stripped.startswith(("{", "<")): # Another JSON file of the export, or the history's HTML variant. raise TakeoutError("not_takeout") - else: - parsed = _watch_later(stripped) + try: + parsed = ( + _history(_json_list(stripped)) if stripped.startswith("[") else _watch_later(stripped) + ) + # A file the import cannot read, however it fails: bad JSON (a ValueError, as is a number too + # long to convert), nesting deeper than the decoder recurses, a csv field past csv's limit. + except (ValueError, RecursionError, csv.Error) as exc: + if isinstance(exc, TakeoutError): + raise + raise TakeoutError("not_takeout") from None if not parsed.videos: raise TakeoutError("no_videos") return parsed @@ -181,6 +221,61 @@ def run_target(user_id: int) -> str: return f"user:{user_id}" +def target_title(db: Session, target: str) -> str | None: + """Who an import run's target is, for the admin's run lists: the user's name, else their email.""" + _, _, raw = target.partition("user:") + user = db.get(User, int(raw)) if raw.isdigit() else None + return (user.display_name or user.email) if user is not None else None + + +class TooMany(Exception): + """The upload would take the user past `ROWS_MAX` kept videos.""" + + +def stage(db: Session, user: User, parsed: Parsed) -> None: + """Keep an uploaded file's videos for the import, unless that would take the user past + `ROWS_MAX` (`TooMany`). The caller commits.""" + service.lock_takeout(db, user) # the count and the insert are one user's, not two uploads' each + if ( + queries.takeout_rows_besides(db, user, parsed.kind, parsed.videos) + len(parsed.videos) + > ROWS_MAX + ): + raise TooMany + service.stage_takeout(db, user, parsed.kind, parsed.videos) + + +# How many unknown Watch later videos one user may have looked up on the shared API budget in a day: +# two full Watch later lists (two units a hundred ids). The rest wait for the next day's runs. +_lookups = RateLimiter(max_events=2 * WATCH_LATER_MAX, window_seconds=24 * 3600) + + +def _put_in_watch_later(db: Session, user: User) -> dict: + """Put the user's waiting Watch later videos into theirs here: the ones the catalog knows at once, + the others once YouTube was asked for them — when a YouTube-connected account can ask, today's + budget is above the backfill reserve, and the user's own daily share of lookups allows it. A video + not asked about yet waits for a later run. The lookup commits what it fetched; the caller commits + the rest.""" + waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER) + if not waiting: + return {} + ids = [vid for _, vid in waiting] + known = queries.known_video_ids(db, ids) + ask = [vid for vid in ids if vid not in known][:WATCH_LATER_MAX] + asked: set[str] = set() + fetcher = get_service_user(db) if ask else None + if ( + fetcher is not None + and not quota.backfill_reserve_reached() + and _lookups.allow(str(user.id), cost=len(ask)) + ): + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP): + with YouTubeClient(db, fetcher) as yt: + ensure_videos(db, yt, ask) + asked = set(ask) + # Only the rows read above: one a second upload staged meanwhile was never asked about. + return service.apply_takeout_watch_later(db, user, waiting, asked) + + 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 @@ -191,15 +286,8 @@ def run_import(user_id: int) -> dict: return {"skipped": "no user"} result = service.apply_takeout_watched(db, user) db.commit() - waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER) - if waiting: - fetcher = get_service_user(db) - if fetcher is not None: - with quota.attribute(user_id, quota.QuotaAction.VIDEOS_LOOKUP): - with YouTubeClient(db, fetcher) as yt: - ensure_videos(db, yt, [vid for _, vid in waiting]) - result |= service.apply_takeout_watch_later(db, user) - db.commit() + result |= _put_in_watch_later(db, user) + db.commit() outcomes = queries.takeout_outcomes(db, user) result["waiting"] = outcomes.get((TakeoutKind.WATCHED, None), 0) if outcomes.get((TakeoutKind.WATCH_LATER, None)): @@ -208,17 +296,28 @@ def run_import(user_id: int) -> dict: 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: + """Settle what every user's import left waiting and can be settled now, one commit each: a + watched video the catalog has come to know and the Shorts probe has cleared (by its job or a + search's), and the Watch later videos a run could not put in yet. Run after each run of the + `shorts` job, which is what clears the most of them. Returns its counts, or an empty dict when + nothing was waiting that could move.""" + marked = added = 0 + ready = set(queries.takeout_users_ready(db)) + later = set(queries.takeout_users_watch_later_waiting(db)) + for user_id in sorted(ready | later): 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 {} + if user_id in ready: + marked += service.apply_takeout_watched(db, user)["watched"] + db.commit() + if user_id in later: + added += _put_in_watch_later(db, user).get("watch_later_added", 0) + db.commit() + out = {} + if ready: + out["takeout_watched"] = marked + if added: + out["takeout_watch_later_added"] = added + return out diff --git a/backend/app/routes/_common.py b/backend/app/routes/_common.py index 32e75e53..6f904dad 100644 --- a/backend/app/routes/_common.py +++ b/backend/app/routes/_common.py @@ -124,6 +124,14 @@ async def capped_body(request: Request, limit: int) -> bytes: return bytes(body) +async def drain_body(request: Request) -> None: + """Read the rest of the body to nowhere, before a refusal is raised: answered mid-upload, the + client reports a network failure, not the refusal (`capped_stream_to` says why). `bodylimit` + bounds how much there can be.""" + async for _ in request.stream(): + pass + + class ByteSink(Protocol): """Where `capped_stream_to` writes: a file, or anything that takes the bytes on their way to one.""" diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index d6a41e24..db16714e 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -1,3 +1,5 @@ +import threading +from collections.abc import AsyncIterator from datetime import date from functools import partial @@ -10,12 +12,14 @@ 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 +from app.ratelimit import RateLimiter from app.feedquery import BadCursor, facet_counts, feed_count, feed_page, feed_summary, feed_years from app.models import User from app.reader import queries, service, takeout from app.routes._common import ( ProgressIn, capped_body, + drain_body, kept_position, not_found, youtube_refusal, @@ -305,17 +309,44 @@ def get_video_detail( # --- Google Takeout ------------------------------------------------------------------------------ -async def _takeout_body(request: Request, db: Session = Depends(get_db)) -> bytes: +# One upload at a time per user, and ten an hour: a file is held whole while it is read, so these +# bound what one account can make the process hold. The section uploads its files one after another. +_takeout_limiter = RateLimiter(max_events=10, window_seconds=3600) +_takeout_uploading: set[int] = set() +_takeout_uploading_guard = threading.Lock() + + +async def _takeout_body( + request: Request, user: User = Depends(current_user), db: Session = Depends(get_db) +) -> AsyncIterator[bytes]: # The sign-in check has run, and it was a SELECT: hand the connection back before a file of # megabytes arrives, so a slow upload does not hold one `idle in transaction`. release_read_connection(db) - return await capped_body(request, takeout.MAX_BYTES) + with _takeout_uploading_guard: + busy = user.id in _takeout_uploading + if not busy: + _takeout_uploading.add(user.id) + if busy or not _takeout_limiter.allow(str(user.id)): + if not busy: + with _takeout_uploading_guard: + _takeout_uploading.discard(user.id) + # Read to the end before answering: refused mid-upload, the browser reports a network + # failure instead of the refusal (capped_stream_to says why). + await drain_body(request) + if busy: + raise refused(429, "takeout_busy") + raise refused(429, "takeout_cooldown") + try: + yield await capped_body(request, takeout.MAX_BYTES) + finally: + with _takeout_uploading_guard: + _takeout_uploading.discard(user.id) @router.post("/takeout") def upload_takeout( - # The sign-in check FIRST: dependencies resolve in order, so a stranger's file is refused before - # a byte of it is read. + # The sign-in check FIRST: the body's own dependency needs the signed-in user, so a stranger's + # file is refused before a byte of it is read. user: User = Depends(current_user), body: bytes = Depends(_takeout_body), db: Session = Depends(get_db), @@ -327,8 +358,14 @@ def upload_takeout( except takeout.TakeoutError as exc: if exc.reason == "no_videos": raise refused(422, "takeout_no_videos") from None + if exc.reason == "too_many": + raise refused(422, "takeout_list_too_long", limit=takeout.WATCH_LATER_MAX) from None raise refused(422, "takeout_unreadable") from None - service.stage_takeout(db, user, parsed.kind, parsed.videos) + try: + takeout.stage(db, user, parsed) + except takeout.TooMany: + db.rollback() + raise refused(422, "takeout_too_many", limit=takeout.ROWS_MAX) from None db.commit() scheduler.run_ad_hoc( scheduler.TAKEOUT_IMPORT, @@ -353,8 +390,13 @@ def takeout_status(user: User = Depends(current_user), db: Session = Depends(get """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)) + target = takeout.run_target(user.id) + run = jobruns.last_run(db, scheduler.TAKEOUT_IMPORT, target) return { + # Queued and not started: its run row is not written yet, and the last run is still the + # previous one, so the page waits for it rather than taking that one for this upload's. + "queued": scheduler.ad_hoc_queued(scheduler.TAKEOUT_IMPORT, target), + "max_bytes": takeout.MAX_BYTES, "run": None if run is None else { diff --git a/backend/app/routes/runs.py b/backend/app/routes/runs.py index 2672d802..7f882b27 100644 --- a/backend/app/routes/runs.py +++ b/backend/app/routes/runs.py @@ -3,7 +3,7 @@ Admin-only and read-only — rows leave the table through the retention job (`job_run_retention_days`). Filtered on the server and paged by `_keyset`, like the server log and the audit log beside it on the activity page. A row names what a reader needs without a second -request: the channel a user's ad-hoc run worked on, and how many log lines the run left (only +request: what a user's ad-hoc run worked on (the channel, the importing user), and how many log lines the run left (only WARNING and up are stored, so most runs have none — the count says so before the click). """ @@ -116,8 +116,12 @@ def list_runs( limit, lambda r: r.started_at, ) - channels = {r.target for r in runs if r.job in scheduler.AD_HOC_JOBS and r.target} + channels = {r.target for r in runs if r.job == scheduler.CHANNEL_BACKFILL and r.target} titles = catalog_queries.channel_titles(db, channels) + for target in {r.target for r in runs if r.job == scheduler.TAKEOUT_IMPORT and r.target}: + title = scheduler.ad_hoc_target_title(db, scheduler.TAKEOUT_IMPORT, target) + if title is not None: + titles[target] = title lines: dict[int, int] = {} if runs: counted = db.execute( diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 44d395ac..39dd3797 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -274,7 +274,6 @@ def scheduler_snapshot(db) -> dict: open_runs = _open_run_ids() recent = jobruns.recent_runs(db, [*JOB_INTERVALS, *AD_HOC_JOBS], RECENT_RUNS) open_runs |= _open_run_ids() - going_id = max((r for r, j in open_runs.items() if j == CHANNEL_BACKFILL), default=None) jobs = [] for job_id, default_interval in JOB_INTERVALS.items(): a = acts.get(job_id, {}) @@ -290,30 +289,11 @@ def scheduler_snapshot(db) -> dict: ) # Work a user starts on one target: no interval, no "run now" (it needs its target), but its # runs are shown beside the jobs' — it spends the same quota. Its one in-memory guard is the - # target's own (sync/runner), so "running" is the runs this process has open: two channels can - # backfill at once, so a run still going counts even when a newer one already ended. That run then - # leads the row, so its channel, its "no outcome yet" and its dot are one run's, never the - # newer run's outcome under the older one's channel. - backfills = recent.get(CHANNEL_BACKFILL, []) - going = db.get(JobRun, going_id) if going_id is not None else None - if going is not None and going.status != RunStatus.RUNNING: - going = None # its outcome is written; only the bookkeeping has not caught up - if going is not None: - backfills = [going, *(r for r in backfills if r.id != going.id)][:RECENT_RUNS] - shown = backfills[0] if backfills else None - ad_hoc = [ - { - "id": CHANNEL_BACKFILL, - "running": going is not None, - "target": shown.target if shown else None, - "target_title": ( - catalog_queries.channel_title(db, shown.target) - if shown is not None and shown.target - else None - ), - **_last_runs(backfills, open_runs), - } - ] + # target's own (sync/runner, the import's per-user lock), so "running" is the runs this process + # has open: two targets can run at once, so a run still going counts even when a newer one already + # ended. That run then leads the row, so its target, its "no outcome yet" and its dot are one + # run's, never the newer run's outcome under the older one's target. + ad_hoc = [_ad_hoc_row(db, job, recent.get(job, []), open_runs) for job in AD_HOC_JOBS] return { "running_here": running_here, "enabled": settings.scheduler_enabled, @@ -335,6 +315,27 @@ def _seen(run, open_runs: dict[int, str]) -> RunStatus: return RunStatus(run.status) +def _ad_hoc_row(db, job: str, runs: list[JobRun], open_runs: dict[int, str]) -> dict: + going_id = max((r for r, j in open_runs.items() if j == job), default=None) + going = db.get(JobRun, going_id) if going_id is not None else None + if going is not None and going.status != RunStatus.RUNNING: + going = None # its outcome is written; only the bookkeeping has not caught up + if going is not None: + runs = [going, *(r for r in runs if r.id != going.id)][:RECENT_RUNS] + shown = runs[0] if runs else None + return { + "id": job, + "running": going is not None, + "target": shown.target if shown else None, + "target_title": ( + ad_hoc_target_title(db, job, shown.target) + if shown is not None and shown.target + else None + ), + **_last_runs(runs, open_runs), + } + + def _last_runs(runs: list, open_runs: dict[int, str]) -> dict: """A row's run fields from its newest runs (newest first): how the last one went, and the few before it for the history dots.""" @@ -619,6 +620,29 @@ CHANNEL_BACKFILL = "channel_backfill" TAKEOUT_IMPORT = "takeout_import" # Every ad-hoc job id: the page names each one (a frontend contract test reads this tuple). AD_HOC_JOBS = (CHANNEL_BACKFILL, TAKEOUT_IMPORT) +# What each one's target is called on the admin's pages: a channel's title, an importing user's name. +_TARGET_TITLES: dict[str, Callable[..., str | None]] = { + CHANNEL_BACKFILL: catalog_queries.channel_title, + TAKEOUT_IMPORT: takeout.target_title, +} + + +def ad_hoc_target_title(db, job: str, target: str) -> str | None: + """The name of an ad-hoc run's target (None for a job with none, or a target since gone).""" + title = _TARGET_TITLES.get(job) + return title(db, target) if title is not None else None + + +# The ad-hoc runs added and not yet recorded, `job@target`: between the two the target's newest +# run row is still the previous run's, so a page watching it asks `ad_hoc_queued` too. +_queued: set[str] = set() +_queued_guard = threading.Lock() + + +def ad_hoc_queued(job: str, target: str) -> bool: + """Whether a run of `job` on `target` is queued and its run row not written yet.""" + with _queued_guard: + return f"{job}@{target}" in _queued def run_ad_hoc( @@ -634,17 +658,24 @@ def run_ad_hoc( if sched is None: # Only between shutdown and exit: start_scheduler runs at every API startup. raise RuntimeError("the scheduler is not running") + key = f"{job}@{target}" + with _queued_guard: + _queued.add(key) try: sched.add_job( _ad_hoc, "date", args=(job, fn, actor_id, target, notify), - id=f"{job}@{target}", + id=key, misfire_grace_time=None, ) # silent-ok: not a failure — the run for this target is already queued, and it does this work. except ConflictingIdError: log.info("ad-hoc %s for %s already queued", job, target) + except Exception: + with _queued_guard: + _queued.discard(key) + raise def _ad_hoc( @@ -656,6 +687,10 @@ def _ad_hoc( # No row to tell it in: the database is what failed. The log is the record. log.exception("ad-hoc %s for %s not started: its run could not be recorded", job, target) return + finally: + # Recorded (or never will be): the run row now speaks for it. + with _queued_guard: + _queued.discard(f"{job}@{target}") _opened(run_id, job) try: _ad_hoc_work(job, fn, run_id, actor_id if notify else None) @@ -672,7 +707,11 @@ def _ad_hoc_work(job: str, fn: Callable[[], dict], run_id: int, notify_id: int | log.exception("job %s failed", job, extra=logs.event("scheduler.job.failed", job=job)) err, failure = _failure(exc) _close(run_id, RunStatus.ERROR, error=err, failure=failure) - _notify_ad_hoc(notify_id, job, RunStatus.ERROR, err, failure) + # The user's notice names the failure, never the upstream's own text: that is an English + # diagnostic, for the admin's run row and the log, not a user's inbox. + _notify_ad_hoc( + notify_id, job, RunStatus.ERROR, None, failure or {"failure": "unexpected"} + ) return summary, data = _finish_ok(job, run_id, result) _notify_ad_hoc(notify_id, job, RunStatus.OK, summary, data) @@ -725,3 +764,5 @@ def shutdown_scheduler() -> None: if _scheduler is not None: _scheduler.shutdown(wait=False) _scheduler = None + with _queued_guard: + _queued.clear() # what it held will not run diff --git a/backend/tests/test_activity_reads.py b/backend/tests/test_activity_reads.py index c4df1b02..59efed36 100644 --- a/backend/tests/test_activity_reads.py +++ b/backend/tests/test_activity_reads.py @@ -159,6 +159,15 @@ def test_the_run_list_filters_pages_and_names_the_channel_and_the_log_lines(db, assert facets["total"] == 4 +def test_the_run_list_names_the_user_a_takeout_import_ran_for(db, make_user, as_user): + admin = make_user(role="admin") + importer = make_user("importer@example.com") + imported = _run(db, "takeout_import", "ok", importer.id, T0, f"user:{importer.id}", "ad_hoc") + db.commit() + (row,) = as_user(admin).get("/api/admin/runs").json()["items"] + assert (row["id"], row["target_title"]) == (imported, "importer@example.com") + + def test_the_quota_breakdown_covers_its_range_and_one_account(db, make_user, as_user): admin = make_user(role="admin") now = datetime.now(timezone.utc) diff --git a/backend/tests/test_body_limit.py b/backend/tests/test_body_limit.py index ad1f72c2..f2239d2a 100644 --- a/backend/tests/test_body_limit.py +++ b/backend/tests/test_body_limit.py @@ -23,6 +23,7 @@ from app import bodylimit, debugsession, sysconfig from app.config import settings from app.models import SystemConfig from app.bodylimit import DEFAULT_MAX_BODY_BYTES, BodyLimitMiddleware +from app.reader import takeout from app.routes import diagnostics as diagnostics_routes SMALL = b"x" * 16 @@ -369,6 +370,9 @@ def test_every_body_reading_site_is_known_and_fits_under_its_ceiling(): # The upload site's limit is the sysconfig knob, which is bounded BY the edge ceiling rather # than being a constant to compare against — that pairing is its own test below. assert ("POST", "/api/files/upload") in bodylimit.ROUTE_MAX_BODY_BYTES + # A Takeout file is larger than the default: without its ceiling every real history is a 413. + assert takeout.MAX_BYTES > DEFAULT_MAX_BODY_BYTES + assert bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/takeout")] == takeout.MAX_BYTES def test_release_read_connection_ends_the_transaction_it_was_called_for(db): diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index 8f795b46..d3fa418e 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -342,6 +342,29 @@ def test_an_ad_hoc_run_is_the_users_run_of_its_target_with_its_spend_and_failure assert (failed.status, failed.error) == ("error", "quota exceeded") +def _ad_hoc_row(db, job: str) -> dict: + rows = scheduler.scheduler_snapshot(db)["ad_hoc"] + assert [r["id"] for r in rows] == list(scheduler.AD_HOC_JOBS) # every one has its row + return next(r for r in rows if r["id"] == job) + + +def test_a_takeout_import_has_its_row_named_by_the_importing_user(db, make_user): + user = make_user("importer@example.com") + user.display_name = "Ida Importer" + db.commit() + assert _ad_hoc_row(db, "takeout_import")["recent"] == [] + scheduler._ad_hoc("takeout_import", lambda: {"watched": 2}, user.id, f"user:{user.id}") + + row = _ad_hoc_row(db, "takeout_import") + assert (row["running"], row["target"], row["target_title"], row["status"]) == ( + False, + f"user:{user.id}", + "Ida Importer", + "ok", + ) + assert _ad_hoc_row(db, "channel_backfill")["recent"] == [] + + def test_the_page_shows_the_ad_hoc_runs_beside_the_jobs_with_their_channel(db, make_user): user = make_user("shown@example.com") db.add(Channel(id="UC_shown", title="BBC Earth")) @@ -349,7 +372,7 @@ def test_the_page_shows_the_ad_hoc_runs_beside_the_jobs_with_their_channel(db, m scheduler._ad_hoc("channel_backfill", lambda: {"skipped": "at capacity"}, user.id, "UC_none") scheduler._ad_hoc("channel_backfill", lambda: {"videos_added": 69}, user.id, "UC_shown") - (row,) = scheduler.scheduler_snapshot(db)["ad_hoc"] + row = _ad_hoc_row(db, "channel_backfill") assert (row["id"], row["running"], row["target"], row["target_title"]) == ( "channel_backfill", False, @@ -367,12 +390,12 @@ def test_the_page_shows_the_ad_hoc_runs_beside_the_jobs_with_their_channel(db, m seen = [] scheduler._ad_hoc( "channel_backfill", - lambda: seen.append(scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"]) or {}, + lambda: seen.append(_ad_hoc_row(db, "channel_backfill")["running"]) or {}, user.id, "UC_shown", ) assert seen == [True] - assert scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"] is False + assert _ad_hoc_row(db, "channel_backfill")["running"] is False def _finished(db, job) -> list[JobRun]: @@ -397,6 +420,9 @@ def test_a_second_ad_hoc_request_before_the_first_starts_is_that_same_run( "channel_backfill@UC_x", "channel_backfill@UC_y", ] + # Queued, so its target's newest run row is not this run's yet: a page watching it can tell. + assert scheduler.ad_hoc_queued(scheduler.CHANNEL_BACKFILL, "UC_x") + assert not scheduler.ad_hoc_queued(scheduler.CHANNEL_BACKFILL, "UC_z") live_scheduler.remove_all_jobs() # never run: a run finishing after the test would land in the next @@ -475,7 +501,7 @@ def test_a_run_whose_outcome_cannot_be_written_is_not_called_failed(db, make_use scheduler._ad_hoc("channel_backfill", lambda: {"videos_added": 1}, user.id, "UC_blip") assert calls == ["ok"] # Its row is still `running`, but nothing here is: the page does not call it going. - (row,) = scheduler.scheduler_snapshot(db)["ad_hoc"] + row = _ad_hoc_row(db, "channel_backfill") assert (row["running"], row["status"], row["recent"][0]["status"]) == ( False, "interrupted", @@ -529,7 +555,7 @@ def test_an_ad_hoc_run_still_going_shows_although_a_newer_one_ended(db, make_use return {} scheduler._ad_hoc("channel_backfill", long_one, user.id, "UC_long") - (row,) = rows + (row,) = [r for r in rows if r["id"] == "channel_backfill"] assert (row["running"], row["target"], row["target_title"]) == (True, "UC_long", "Long one") # The outcome beside it is the long one's (none yet), never the short one's skip. assert (row["status"], row["result"], row["last_error"]) == (None, None, None) diff --git a/backend/tests/test_takeout.py b/backend/tests/test_takeout.py index f603c306..823b5d3e 100644 --- a/backend/tests/test_takeout.py +++ b/backend/tests/test_takeout.py @@ -3,16 +3,20 @@ import job makes of them — watched here (never a Short, never over a state the a video the catalog does not know yet, and the Watch later list joined to the user's own.""" import json +import tracemalloc from datetime import datetime, timedelta, timezone import pytest -from sqlalchemy import select +from sqlalchemy import select, text +from sqlalchemy.exc import OperationalError -from app import jobruns, scheduler +from app import jobruns, quota, scheduler +from app.db import SessionLocal from app.catalog.models import Channel, Video -from app.models import JobRun, Notification +from app.models import JobRun, Notification, User from app.runstatus import RunStatus, RunTrigger -from app.reader import service, takeout +from app.reader import queries, service, takeout +from app.routes import feed as feed_routes from app.reader.models import ( Playlist, PlaylistItem, @@ -107,6 +111,18 @@ def test_the_watch_later_csv_is_read_with_a_header_in_any_language_or_none(): json.dumps([_view("aaaaaaaaaaa", details=[{"name": "From Google Ads"}])]).encode(), "no_videos", ), + # Read an entry at a time, the list must still be one: nothing after it, nothing missing. + (_history(_view("aaaaaaaaaaa")) + b" []", "not_takeout"), + (_history(_view("aaaaaaaaaaa"))[:-1], "not_takeout"), + (b"[" + json.dumps(_view("aaaaaaaaaaa")).encode() + b" x]", "not_takeout"), + # However a hostile file fails, it is one the import cannot read, not a server error. + (b"[" * 100_000, "not_takeout"), + (b"[" + b"1" * 5000 + b"]", "not_takeout"), + (b"Ngb04xUK8uU," + b"x" * 200_000 + b"\n", "not_takeout"), + ( + "".join(f"v{n:010d},2019-02-07T15:23:30Z\n" for n in range(5001)).encode(), + "too_many", + ), ], ) def test_a_file_the_import_cannot_use_is_refused_with_its_reason(body, reason): @@ -115,6 +131,42 @@ def test_a_file_the_import_cannot_use_is_refused_with_its_reason(body, reason): assert caught.value.reason == reason +def test_the_history_is_read_an_entry_at_a_time_not_as_one_tree(): + """A file of a million tiny entries must cost about its own size, not a million dicts at once: + the whole tree built first measured ~25 times the file.""" + body = b"[" + b",".join([b"{}"] * 1_000_000) + b"]" # 3 MB + tracemalloc.start() + try: + with pytest.raises(takeout.TakeoutError): + takeout.parse(body) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert peak < 5 * len(body), f"parsing held {peak / len(body):.1f} times the file" + + +def test_the_id_lists_are_asked_in_chunks(db, make_user, monkeypatch): + user = make_user("chunks@example.com") + ids = [f"chunk{n:06d}" for n in range(5)] + for vid in ids[:3]: + _video(db, vid) + db.commit() + service.stage_takeout(db, user, TakeoutKind.WATCHED, {vid: (T0, None) for vid in ids[:4]}) + db.commit() + monkeypatch.setattr(queries, "_IN_CHUNK", 2) + assert queries.known_video_ids(db, ids) == set(ids[:3]) + assert queries.takeout_rows_besides(db, user, TakeoutKind.WATCHED, ids) == 0 + assert queries.takeout_rows_besides(db, user, TakeoutKind.WATCHED, ids[:1]) == 3 + + +def test_the_history_is_read_whatever_its_whitespace(): + body = b" \n[ \n" + json.dumps(_view("aaaaaaaaaaa")).encode() + b" ,\n\t" + body += json.dumps(_view("bbbbbbbbbbb")).encode() + b"\n] \n" + assert sorted(takeout.parse(body).videos) == ["aaaaaaaaaaa", "bbbbbbbbbbb"] + full = "".join(f"v{n:010d},2019-02-07T15:23:30Z\n" for n in range(5000)) + assert len(takeout.parse(full.encode()).videos) == 5000 # YouTube's own cap still fits + + # --- keeping and applying ---------------------------------------------------------------------- @@ -234,10 +286,15 @@ def test_the_watch_later_list_joins_the_users_own_oldest_first_after_what_it_hol }, ) - counts = service.apply_takeout_watch_later(db, user) + service.stage_takeout(db, user, TakeoutKind.WATCH_LATER, {"unasked0000": (T0, None)}) + rows = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER) + + counts = service.apply_takeout_watch_later(db, user, rows, asked={"gone0000000"}) db.commit() assert counts == {"watch_later_added": 2, "watch_later_present": 1, "watch_later_missing": 1} + # Unknown and not asked about: it waits, it is not called gone. + assert _rows(db, user)["WL:unasked0000"].outcome is None items = db.execute( select(PlaylistItem.video_id) .where(PlaylistItem.playlist_id == pl.id) @@ -265,16 +322,17 @@ def test_the_job_fetches_the_unknown_watch_later_videos_and_says_what_waits( asked = [] def fake_ensure(session, yt, ids): - asked.append(ids) + # The lookup is the importing user's spend, on the import's own action. + asked.append((ids, quota._actor_id.get(), quota._action.get())) _video(session, "fetched0000") - session.commit() + session.flush() monkeypatch.setattr(takeout, "ensure_videos", fake_ensure) monkeypatch.setattr(takeout, "get_service_user", lambda session: user) result = takeout.run_import(user.id) - assert asked == [["fetched0000"]] + assert asked == [(["fetched0000"], user.id, quota.QuotaAction.VIDEOS_LOOKUP)] assert result == { "watched": 1, "kept": 0, @@ -286,13 +344,152 @@ def test_the_job_fetches_the_unknown_watch_later_videos_and_says_what_waits( } +def _wl_job(db, make_user, monkeypatch, email, *, fetcher=True, reserve=False): + user = make_user(email) + _video(db, "known000000") + db.commit() + service.stage_takeout( + db, + user, + TakeoutKind.WATCH_LATER, + {"known000000": (T0, None), "unknown0000": (T0 + timedelta(minutes=1), None)}, + ) + db.commit() + asked = [] + monkeypatch.setattr(takeout, "ensure_videos", lambda session, yt, ids: asked.append(ids)) + monkeypatch.setattr(takeout, "get_service_user", lambda session: user if fetcher else None) + monkeypatch.setattr(takeout.quota, "backfill_reserve_reached", lambda: reserve) + return user, asked + + +@pytest.mark.parametrize( + "why", ["no account to ask with", "the budget's reserve", "the user's share"] +) +def test_a_watch_later_video_not_asked_about_waits_and_the_known_ones_go_in( + db, make_user, monkeypatch, why +): + user, asked = _wl_job( + db, + make_user, + monkeypatch, + "wait@example.com", + fetcher=why != "no account to ask with", + reserve=why == "the budget's reserve", + ) + if why == "the user's share": + monkeypatch.setattr(takeout, "_lookups", takeout.RateLimiter(0, 3600)) + + result = takeout.run_import(user.id) + + assert asked == [] + assert (result["watch_later_added"], result["watch_later_missing"]) == (1, 0) + assert result["watch_later_waiting"] == 1 + assert _rows(db, user)["WL:unknown0000"].outcome is None + + +def test_the_shorts_job_puts_in_the_watch_later_videos_a_run_could_not(db, make_user, monkeypatch): + user, asked = _wl_job(db, make_user, monkeypatch, "retry@example.com", fetcher=False) + takeout.run_import(user.id) + assert asked == [] + + # A YouTube-connected account exists now; the next probe run asks for what waits. + monkeypatch.setattr(takeout, "get_service_user", lambda session: user) + + def fetched(session, yt, ids): + asked.append(ids) + _video(session, "unknown0000") + session.flush() + + monkeypatch.setattr(takeout, "ensure_videos", fetched) + assert takeout.apply_waiting(db) == {"takeout_watch_later_added": 1} + assert asked == [["unknown0000"]] + assert _rows(db, user)["WL:unknown0000"].outcome == "added" + assert takeout.apply_waiting(db) == {} + + +def test_a_watch_later_row_staged_while_youtube_is_asked_is_left_for_the_next_run( + db, make_user, monkeypatch +): + user, _ = _wl_job(db, make_user, monkeypatch, "race@example.com") + + def meanwhile(session, yt, ids): + # A second upload lands while the lookup runs: its video was never asked about. + with SessionLocal() as other: + service.stage_takeout( + other, + other.get(User, user.id), + TakeoutKind.WATCH_LATER, + {"second00000": (T0, None)}, + ) + other.commit() + + monkeypatch.setattr(takeout, "ensure_videos", meanwhile) + takeout.run_import(user.id) + + rows = _rows(db, user) + assert rows["WL:unknown0000"].outcome == "missing" # asked, and YouTube has it no longer + assert rows["WL:second00000"].outcome is None + + +def test_a_youtube_failure_keeps_what_the_watched_part_already_marked(db, make_user, monkeypatch): + user = make_user("fails@example.com") + _video(db, "plain000000") + db.commit() + service.stage_takeout(db, user, TakeoutKind.WATCHED, {"plain000000": (T0, CHANNEL)}) + service.stage_takeout(db, user, TakeoutKind.WATCH_LATER, {"unknown0000": (T0, None)}) + db.commit() + + def refused(session, yt, ids): + raise RuntimeError("quota exceeded") + + monkeypatch.setattr(takeout, "ensure_videos", refused) + monkeypatch.setattr(takeout, "get_service_user", lambda session: user) + with pytest.raises(RuntimeError): + takeout.run_import(user.id) + + with SessionLocal() as fresh: + state = fresh.execute(select(VideoState).where(VideoState.user_id == user.id)).scalar_one() + assert (state.video_id, state.status) == ("plain000000", "watched") + outcome = fresh.execute( + select(TakeoutItem.outcome).where(TakeoutItem.video_id == "plain000000") + ).scalar_one() + assert outcome == "watched" + + +def test_staging_and_settling_take_the_users_one_lock(db, make_user): + """An upload's staging and an import's settling lock the same rows in no common order: one lock + over both is what keeps the two from deadlocking.""" + user = make_user("lock@example.com") + db.commit() + steps = { + "stage": lambda s, u: service.stage_takeout( + s, u, TakeoutKind.WATCHED, {"aaaaaaaaaaa": (T0, None)} + ), + "watched": lambda s, u: service.apply_takeout_watched(s, u), + "watch later": lambda s, u: service.apply_takeout_watch_later( + s, u, [(1, "aaaaaaaaaaa")], set() + ), + } + with SessionLocal() as holder: + service.lock_takeout(holder, holder.get(User, user.id)) + for name, step in steps.items(): + with SessionLocal() as other: + other.execute(text("SET lock_timeout = '200ms'")) + with pytest.raises(OperationalError, match="lock timeout"): + step(other, other.get(User, user.id)) + pytest.fail(f"{name} did not wait for the user's lock") + holder.rollback() + + # --- the upload -------------------------------------------------------------------------------- @pytest.fixture def queued(monkeypatch): runs = [] - monkeypatch.setattr(scheduler, "run_ad_hoc", lambda job, fn, **kw: runs.append((job, kw))) + monkeypatch.setattr( + scheduler, "run_ad_hoc", lambda job, fn, **kw: runs.append((job, fn.func, fn.args, kw)) + ) return runs @@ -318,6 +515,8 @@ def test_an_upload_keeps_its_videos_and_queues_the_users_import_with_a_notice( assert queued == [ ( "takeout_import", + takeout.run_import, + (user.id,), {"actor_id": user.id, "target": f"user:{user.id}", "notify": True}, ) ] @@ -345,8 +544,87 @@ def test_an_upload_over_the_cap_is_refused_while_it_arrives(db, make_user, as_us assert _rows(db, user) == {} -def test_an_anonymous_upload_is_refused(client): - assert client.post("/api/takeout", content=_history(_view("aaaaaaaaaaa"))).status_code == 401 +def _counting_body(chunks: int = 4, size: int = 1024): + """A body that records how much of itself the server pulled (test_upload.py's probe).""" + pulled: list[int] = [] + + def body(): + for _ in range(chunks): + pulled.append(size) + yield b"x" * size + + return body(), pulled + + +def test_an_anonymous_upload_is_refused_before_a_byte_of_it_is_read(client): + body, pulled = _counting_body() + assert client.post("/api/takeout", content=body).status_code == 401 + assert pulled == [], f"the server read {sum(pulled)} bytes from an anonymous caller" + + +def test_a_signed_in_upload_has_its_body_read(db, make_user, as_user, queued): + """The mutation check for the test above: a route reading nobody's body would pass it too.""" + body, pulled = _counting_body() + c = as_user(make_user("read@example.com")) + assert c.post("/api/takeout", content=body).status_code == 422 + assert len(pulled) == 4 + + +def test_uploads_past_the_hourly_limit_are_refused_after_their_body_arrived( + db, make_user, as_user, queued +): + c = as_user(make_user("many@example.com")) + for _ in range(10): + assert c.post("/api/takeout", content=_history(_view("aaaaaaaaaaa"))).status_code == 200 + body, pulled = _counting_body() + resp = c.post("/api/takeout", content=body) + assert resp.status_code == 429 + assert resp.json()["detail"] == {"code": "refused", "reason": "takeout_cooldown"} + # Read to the end first: refused mid-upload, the browser would say "network error" instead. + assert len(pulled) == 4 + assert len(queued) == 10 + + +def test_a_second_upload_while_one_of_the_users_is_still_arriving_is_refused( + db, make_user, as_user, queued, monkeypatch +): + user = make_user("busy@example.com") + monkeypatch.setattr(feed_routes, "_takeout_uploading", {user.id}) + resp = as_user(user).post("/api/takeout", content=_history(_view("aaaaaaaaaaa"))) + assert resp.status_code == 429 + assert resp.json()["detail"]["reason"] == "takeout_busy" + assert queued == [] + # Another user's upload is not held up by it, and lets go of its own place when it is done. + other = as_user(make_user("free@example.com")) + assert other.post("/api/takeout", content=_history(_view("aaaaaaaaaaa"))).status_code == 200 + assert feed_routes._takeout_uploading == {user.id} + + +def test_an_upload_that_would_keep_too_many_videos_is_refused( + db, make_user, as_user, queued, monkeypatch +): + user = make_user("cap@example.com") + c = as_user(user) + monkeypatch.setattr(takeout, "ROWS_MAX", 3) + two = _history(_view("aaaaaaaaaaa"), _view("bbbbbbbbbbb")) + assert c.post("/api/takeout", content=two).status_code == 200 + assert c.post("/api/takeout", content=two).status_code == 200 # the same videos again: no more + resp = c.post("/api/takeout", content=_history(_view("ccccccccccc"), _view("ddddddddddd"))) + assert resp.status_code == 422 + assert resp.json()["detail"] == {"code": "refused", "reason": "takeout_too_many", "limit": 3} + assert sorted(_rows(db, user)) == ["aaaaaaaaaaa", "bbbbbbbbbbb"] + assert len(queued) == 2 + + +def test_a_watch_later_file_longer_than_youtube_allows_is_refused(db, make_user, as_user, queued): + body = "".join(f"v{n:010d},2019-02-07T15:23:30Z\n" for n in range(5001)).encode() + resp = as_user(make_user("long@example.com")).post("/api/takeout", content=body) + assert resp.status_code == 422 + assert resp.json()["detail"] == { + "code": "refused", + "reason": "takeout_list_too_long", + "limit": 5000, + } # --- the notice -------------------------------------------------------------------------------- @@ -363,6 +641,50 @@ def test_an_ad_hoc_run_that_asks_for_it_tells_its_user_how_it_went(db, make_user ] +def test_a_failed_run_tells_its_user_the_failure_never_the_upstreams_text(db, make_user): + user = make_user("failnotice@example.com") + + def fails(): + raise RuntimeError("(psycopg.errors.Something) [SQL: SELECT secret]") + + scheduler._ad_hoc("takeout_import", fails, user.id, "user:1", True) + scheduler._ad_hoc("channel_backfill", fails, user.id, "UC_x") + db.expire_all() + (notice,) = db.execute(select(Notification).where(Notification.user_id == user.id)).scalars() + assert notice.body is None + assert notice.data == { + "job_id": "takeout_import", + "status": "error", + "summary": None, + "result": {"failure": "unexpected"}, + } + # The admin's run row keeps the diagnostic. + run = db.execute(select(JobRun).where(JobRun.job == "takeout_import")).scalar_one() + assert "SELECT secret" in run.error + + +def test_a_queued_import_is_told_apart_from_the_last_one_until_its_run_is_recorded( + db, make_user, as_user, monkeypatch +): + user = make_user("queued@example.com") + added = [] + + class Held: + def add_job(self, *args, **kwargs): + added.append(kwargs["id"]) + + monkeypatch.setattr(scheduler, "_scheduler", Held()) + target = takeout.run_target(user.id) + scheduler.run_ad_hoc("takeout_import", dict, actor_id=user.id, target=target) + assert scheduler.ad_hoc_queued("takeout_import", target) + assert as_user(user).get("/api/takeout").json()["queued"] is True + + scheduler._ad_hoc("takeout_import", dict, user.id, target) + assert not scheduler.ad_hoc_queued("takeout_import", target) + assert as_user(user).get("/api/takeout").json()["queued"] is False + assert added == [f"takeout_import@{target}"] + + # --- the videos that wait ---------------------------------------------------------------------- @@ -410,6 +732,8 @@ def test_the_shorts_job_settles_the_waiting_rows_its_probe_made_ready(db, make_u 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() == { + "queued": False, + "max_bytes": takeout.MAX_BYTES, "run": None, "watched": {}, "watch_later": {}, diff --git a/frontend/e2e/global.setup.ts b/frontend/e2e/global.setup.ts index a7942a82..2c68e31a 100644 --- a/frontend/e2e/global.setup.ts +++ b/frontend/e2e/global.setup.ts @@ -1,9 +1,9 @@ import { test as setup, expect, type Page } from "@playwright/test"; import { execFileSync } from "node:child_process"; -import { mkdirSync, readFileSync } from "node:fs"; +import { mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; -import { MEMBER_STORAGE } from "./helpers"; +import { MEMBER_STORAGE, SEED_FILE } from "./helpers"; import { retryConnect } from "./retryConnect"; /** @@ -99,6 +99,8 @@ setup("seed + authenticate", async ({ page, browser }) => { { input: readFileSync(SEED_SCRIPT) }, ); console.log(`[e2e seed] ${out.toString().trim()}`); + // What the seed chose, for the specs that need one of its videos (helpers.seeded). + writeFileSync(SEED_FILE, out); } catch (err: unknown) { const e = err as { stderr?: Buffer; message?: string }; throw new Error(`E2E seed failed: ${e.stderr?.toString() || e.message}`); diff --git a/frontend/e2e/helpers.ts b/frontend/e2e/helpers.ts index 8abc629c..e4ddef32 100644 --- a/frontend/e2e/helpers.ts +++ b/frontend/e2e/helpers.ts @@ -1,8 +1,18 @@ +import { readFileSync } from "node:fs"; import { expect, type Page } from "@playwright/test"; /** The plain member's saved session (global.setup.ts signs it in); `member.spec.ts` and * `devices.spec.ts` run from it. */ export const MEMBER_STORAGE = "e2e/.auth/member.json"; +// The seed's own answer (scripts/e2e_seed.py), written by global.setup: the videos it chose. +export const SEED_FILE = "e2e/.auth/seed.json"; + +/** The videos the seed chose for the main account under `key` in its answer. */ +export function seeded(key: string): string[] { + const ids = (JSON.parse(readFileSync(SEED_FILE, "utf-8")) as Record)[key]; + if (!Array.isArray(ids) || !ids.length) throw new Error(`the seed chose no ${key} videos`); + return ids as string[]; +} /** Open the app and wait for the authenticated shell (the nav rail) to be ready. */ export async function openApp(page: Page, path = "/"): Promise { diff --git a/frontend/e2e/takeout.spec.ts b/frontend/e2e/takeout.spec.ts index 1794f171..a42c0d25 100644 --- a/frontend/e2e/takeout.spec.ts +++ b/frontend/e2e/takeout.spec.ts @@ -1,20 +1,16 @@ import { test, expect } from "@playwright/test"; -import { openApp, gotoPage } from "./helpers"; +import { openApp, gotoPage, seeded } from "./helpers"; // Settings › Account › YouTube history: a Google Takeout watch history, uploaded through the real // picker, is read, imported in the background, and the section says what it did — a video the // catalog knows is marked watched, one it does not know is remembered for later. The seed clears // the account's kept Takeout videos each run, but the projects share one account: a later one's -// feed no longer holds the video an earlier one marked, so what it adds is counted against what was -// there before it. +// import of the same video adds nothing new, so what each adds is counted against what was there +// before it. test("a Takeout watch history uploads, imports, and the section says what it did", async ({ page, }) => { await openApp(page); - // A video the catalog knows: one on the account's feed (not a Short — the feed leaves them out). - const known = await page.getByTestId("video-card").first().getAttribute("data-video-id"); - expect(known).toBeTruthy(); - await gotoPage(page, "settings"); await page.getByRole("tab", { name: "Account" }).click(); await expect(page.getByRole("heading", { name: "YouTube history" })).toBeVisible(); @@ -28,6 +24,10 @@ test("a Takeout watch history uploads, imports, and the section says what it did watched?: Record; }; const watchedBefore = before.watched?.watched ?? 0; + // A video the catalog knows and the Shorts probe has cleared, chosen by the seed (one not probed + // yet would wait for the probe): the next one no earlier project of this run has imported. + const known = seeded("takeout_known")[watchedBefore]; + expect(known, "the seed chose too few videos for every project").toBeTruthy(); // Watched now: later than any state the seed set, which the import would otherwise leave alone. const now = new Date().toISOString(); diff --git a/frontend/src/components/ui/DropZone.tsx b/frontend/src/components/ui/DropZone.tsx index ada39182..65a6130c 100644 --- a/frontend/src/components/ui/DropZone.tsx +++ b/frontend/src/components/ui/DropZone.tsx @@ -1,4 +1,4 @@ -import { useRef, useState, type ReactNode } from "react"; +import { useRef, useState, type ReactNode, type Ref } from "react"; import clsx from "clsx"; import { Pressable } from "./Pressable"; @@ -8,6 +8,7 @@ import { Pressable } from "./Pressable"; // children (an icon, a line of text), so a plain boolean flickers — enter and leave are counted // and the highlight clears at depth 0. The caller draws the inside, told whether a drag is over it. export function DropZone({ + ref, onFiles, multiple = false, accept, @@ -17,6 +18,8 @@ export function DropZone({ className, children, }: { + /** The zone's button, for a caller that moves the focus to it. */ + ref?: Ref; /** The files dropped or picked, never empty. One unless `multiple`. */ onFiles: (files: [File, ...File[]]) => void; multiple?: boolean; @@ -40,6 +43,7 @@ export function DropZone({ return ( <> inputRef.current?.click()} onDragEnter={(e) => { diff --git a/frontend/src/i18n/backendIds.contract.test.ts b/frontend/src/i18n/backendIds.contract.test.ts index 9b2496ed..9b08e934 100644 --- a/frontend/src/i18n/backendIds.contract.test.ts +++ b/frontend/src/i18n/backendIds.contract.test.ts @@ -313,6 +313,29 @@ describe("backend ids have their names in both languages", () => { expect(/^TAKEOUT_IMPORT = "(\w+)"$/m.exec(schedulerPy)?.[1]).toBe(ours); }); + it("the Takeout import's files and outcomes are the server's", () => { + const feedTs = readFileSync(new URL("../lib/api/feed.ts", import.meta.url), "utf8"); + const ours = (name: string) => + [ + ...( + new RegExp(`^export const ${name} = \\[([^\\]]*)\\]`, "m").exec(feedTs)?.[1] ?? "" + ).matchAll(/"(\w+)"/g), + ].map((m) => m[1]); + const models = backend("reader/models.py"); + const theirs = (cls: string) => + [ + ...( + new RegExp(`^class ${cls}\\(StrEnum\\):\\n([\\s\\S]*?)(?=\\n\\S)`, "m").exec( + models, + )?.[1] ?? "" + ).matchAll(/^\s+[A-Z_]+ = "(\w+)"/gm), + ].map((m) => m[1]); + expect(theirs("TakeoutKind").length).toBeGreaterThan(0); + expect(ours("TAKEOUT_KINDS")).toEqual(theirs("TakeoutKind")); + expect(theirs("TakeoutOutcome").length).toBeGreaterThan(0); + expect(ours("TAKEOUT_OUTCOMES")).toEqual(theirs("TakeoutOutcome")); + }); + for (const lang of ["en", "hu"]) { const keys = keysOf(lang); const missing = (wanted: string[]) => wanted.filter((k) => !keys.has(k)); diff --git a/frontend/src/i18n/locales/en/errors.json b/frontend/src/i18n/locales/en/errors.json index be7692c4..c9066502 100644 --- a/frontend/src/i18n/locales/en/errors.json +++ b/frontend/src/i18n/locales/en/errors.json @@ -252,6 +252,10 @@ "api_token_name_required": "Give the token a name", "api_token_limit": "You already have {{limit}} tokens — revoke one first", "takeout_unreadable": "This isn't a Takeout file the import reads. Upload watch-history.json (JSON format) or Watch later-videos.csv.", - "takeout_no_videos": "This file has no videos to import" + "takeout_no_videos": "This file has no videos to import", + "takeout_busy": "Another of your files is still being uploaded — wait for it to finish", + "takeout_cooldown": "You have uploaded many files in the last hour — try again later", + "takeout_too_many": "That would keep more than {{limit, number}} videos from your export — more than a real one holds", + "takeout_list_too_long": "This list holds more than {{limit, number}} videos — more than a YouTube list can" } } diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index fbd6410b..1eafde92 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -187,7 +187,8 @@ "watch_later_present": "already in Watch later", "watch_later_missing": "no longer on YouTube", "watch_later_waiting": "Watch later waiting", - "takeout_watched": "marked watched from Takeout" + "takeout_watched": "marked watched from Takeout", + "takeout_watch_later_added": "put into Watch later from Takeout" }, "skip": { "disabled": "disabled", @@ -205,7 +206,8 @@ }, "failure": { "rss_all_failed_one": "The RSS feed did not answer for the one channel, {{channel}}: {{cause}}. Read through the API instead: {{read}}.", - "rss_all_failed_other": "The RSS feed answered for none of the {{total}} channels — the first: {{channel}}, {{cause}}. Read through the API instead: {{read}}." + "rss_all_failed_other": "The RSS feed answered for none of the {{total}} channels — the first: {{channel}}, {{cause}}. Read through the API instead: {{read}}.", + "unexpected": "The run ran into an unexpected error." }, "rssCause": { "http": "HTTP {{status}}", diff --git a/frontend/src/i18n/locales/en/settings.json b/frontend/src/i18n/locales/en/settings.json index 6126ebbf..3bf5f3d3 100644 --- a/frontend/src/i18n/locales/en/settings.json +++ b/frontend/src/i18n/locales/en/settings.json @@ -52,7 +52,8 @@ "missing_other": "{{n}} no longer on YouTube", "laterWaiting_one": "{{n}} waiting for Watch later", "laterWaiting_other": "{{n}} waiting for Watch later" - } + }, + "tooLarge": "{{name}} is larger than {{limit}} — the import takes files up to that size" }, "save": { "saving": "Saving…", diff --git a/frontend/src/i18n/locales/hu/errors.json b/frontend/src/i18n/locales/hu/errors.json index cf797edc..6063a8e4 100644 --- a/frontend/src/i18n/locales/hu/errors.json +++ b/frontend/src/i18n/locales/hu/errors.json @@ -252,6 +252,10 @@ "api_token_name_required": "Adj nevet a tokennek", "api_token_limit": "Már {{limit}} tokened van — előbb vonj vissza egyet", "takeout_unreadable": "Ez nem olyan Takeout-fájl, amit az import olvas. A watch-history.json (JSON formátum) vagy a Watch later-videos.csv fájlt töltsd fel.", - "takeout_no_videos": "Ebben a fájlban nincs importálható videó" + "takeout_no_videos": "Ebben a fájlban nincs importálható videó", + "takeout_busy": "Egy másik fájlod feltöltése még folyik — várd meg, amíg befejeződik", + "takeout_cooldown": "Az elmúlt órában sok fájlt töltöttél fel — próbáld újra később", + "takeout_too_many": "Így több mint {{limit, number}} videót őriznénk meg az exportodból — többet, mint amennyi egy valódiban van", + "takeout_list_too_long": "Ez a lista több mint {{limit, number}} videót tartalmaz — többet, mint amennyi egy YouTube-listán lehet" } } diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index 029ad15b..2506ca42 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -187,7 +187,8 @@ "watch_later_present": "már a „Megnézem később” listán", "watch_later_missing": "már nincs a YouTube-on", "watch_later_waiting": "a „Megnézem később” listára vár", - "takeout_watched": "Takeoutból megnézettre jelölve" + "takeout_watched": "Takeoutból megnézettre jelölve", + "takeout_watch_later_added": "Takeoutból a „Megnézem később” listába téve" }, "skip": { "disabled": "kikapcsolva", @@ -205,7 +206,8 @@ }, "failure": { "rss_all_failed_one": "Az RSS-hírcsatorna az egyetlen csatornánál sem válaszolt ({{channel}}): {{cause}}. Az API-n át beolvasva helyette: {{read}}.", - "rss_all_failed_other": "Az RSS-hírcsatorna egyik csatornánál sem válaszolt ({{total}}) — az első: {{channel}}, {{cause}}. Az API-n át beolvasva helyette: {{read}}." + "rss_all_failed_other": "Az RSS-hírcsatorna egyik csatornánál sem válaszolt ({{total}}) — az első: {{channel}}, {{cause}}. Az API-n át beolvasva helyette: {{read}}.", + "unexpected": "A futás váratlan hibába ütközött." }, "rssCause": { "http": "HTTP {{status}}", diff --git a/frontend/src/i18n/locales/hu/settings.json b/frontend/src/i18n/locales/hu/settings.json index be0c1039..44498e47 100644 --- a/frontend/src/i18n/locales/hu/settings.json +++ b/frontend/src/i18n/locales/hu/settings.json @@ -44,15 +44,16 @@ "short_other": "{{n}} Short kihagyva", "kept_one": "{{n}} úgy maradt, ahogy itt beállítottad", "kept_other": "{{n}} úgy maradt, ahogy itt beállítottad", - "added_one": "{{n}} a „Megnézem később” listába", - "added_other": "{{n}} a „Megnézem később” listába", + "added_one": "{{n}} a „Megnézem később” listába téve", + "added_other": "{{n}} a „Megnézem később” listába téve", "present_one": "{{n}} már a „Megnézem később” listán", "present_other": "{{n}} már a „Megnézem később” listán", "missing_one": "{{n}} már nincs fent a YouTube-on", "missing_other": "{{n}} már nincs fent a YouTube-on", "laterWaiting_one": "{{n}} vár a „Megnézem később” listára", "laterWaiting_other": "{{n}} vár a „Megnézem később” listára" - } + }, + "tooLarge": "{{name}}: nagyobb, mint {{limit}} — az import legfeljebb ekkora fájlt fogad" }, "save": { "saving": "Mentés…", diff --git a/frontend/src/lib/api/feed.ts b/frontend/src/lib/api/feed.ts index 0ffbfe83..41a89376 100644 --- a/frontend/src/lib/api/feed.ts +++ b/frontend/src/lib/api/feed.ts @@ -164,12 +164,26 @@ export interface TakeoutUpload { other: number; malformed: number; } -export type TakeoutKind = "watched" | "watch_later"; -// A kept video's outcome (`reader.models.TakeoutOutcome`), or "waiting" while the catalog does not -// know it yet: watched / kept (the user's own state here stood) / short for the history; added / -// present / missing for Watch later. -export type TakeoutCounts = Partial>; +// The two files (`reader.models.TakeoutKind`) and a kept video's outcome (`TakeoutOutcome`): watched +// / kept (the user's own state here stood) / short for the history; added / present / missing for +// Watch later. backendIds.contract.test holds both lists to the backend's enums. +export const TAKEOUT_KINDS = ["watched", "watch_later"] as const; +export const TAKEOUT_OUTCOMES = [ + "watched", + "kept", + "short", + "added", + "present", + "missing", +] as const; +export type TakeoutKind = (typeof TAKEOUT_KINDS)[number]; +// Each outcome's count, and "waiting" for the videos the import could not settle yet. +export type TakeoutCounts = Partial>; export interface TakeoutStatus { + /** An import is queued and its run not started: `run` is still the one before it. */ + queued: boolean; + /** The largest file the upload takes. */ + max_bytes: number; run: { status: RunStatus; started_at: string | null; diff --git a/frontend/src/modules/settings/TakeoutImport.dom.test.tsx b/frontend/src/modules/settings/TakeoutImport.dom.test.tsx index 0675226a..88701b94 100644 --- a/frontend/src/modules/settings/TakeoutImport.dom.test.tsx +++ b/frontend/src/modules/settings/TakeoutImport.dom.test.tsx @@ -56,6 +56,14 @@ const uploaded = (videos: number): TakeoutUpload => ({ malformed: 0, }); +// A whole status: nothing queued and room for any file, unless the test says otherwise. +const st = (s: Partial): TakeoutStatus => ({ + queued: false, + max_bytes: 1_000_000, + run: null, + ...s, +}); + function mount(status: TakeoutStatus | Error | Promise) { const spy = vi.spyOn(api, "takeoutStatus"); if (status instanceof Error) spy.mockRejectedValue(status); @@ -77,11 +85,13 @@ const drop = (files: File[]) => act(async () => { fireEvent.drop(zone()!, { dataTransfer: { files } }); }); -const file = (name: string) => new File(["[]"], name); +const file = (name: string, size = 2) => new File(["x".repeat(size)], name); +const unavailable = (el: HTMLElement | null) => + !!el && ((el as HTMLButtonElement).disabled || el.getAttribute("aria-disabled") === "true"); describe("YouTube history import · what it says", () => { it("before the first import: the drop zone, and no line about an import", async () => { - mount({ run: null, watched: {}, watch_later: {} }); + mount(st({ run: null, watched: {}, watch_later: {} })); await waitFor(() => expect(zone()).toBeTruthy()); expect(screen.queryByTestId("takeout-last")).toBeNull(); }); @@ -92,11 +102,13 @@ describe("YouTube history import · what it says", () => { }); it("after one: when, then only the counts it produced, the history's first", async () => { - mount({ - run: run("ok"), - watched: { watched: 1020, waiting: 6785, short: 0, kept: 81 }, - watch_later: { added: 41, waiting: 2 }, - }); + mount( + st({ + run: run("ok"), + watched: { watched: 1020, waiting: 6785, short: 0, kept: 81 }, + watch_later: { added: 41, waiting: 2 }, + }), + ); await waitFor(() => expect(screen.getByTestId("takeout-last")).toBeTruthy()); expect(line()).toBe( [ @@ -114,20 +126,20 @@ describe("YouTube history import · what it says", () => { }); it("a failed run and one a restart cut off say so, not counts", async () => { - mount({ run: run("error"), watched: { watched: 3 } }); + mount(st({ run: run("error"), watched: { watched: 3 } })); await waitFor(() => expect(line()).toBe("settings.takeout.failed(at F)")); cleanup(); - mount({ run: run("interrupted"), watched: { watched: 3 } }); + mount(st({ run: run("interrupted"), watched: { watched: 3 } })); await waitFor(() => expect(line()).toBe("settings.takeout.interrupted(at S)")); }); - it("a running import: the running line, no drop zone, asked again until it ends", async () => { - const spy = mount({ run: run("running"), watched: {} }); + it("a running import: the running line, the zone unavailable, asked again until it ends", async () => { + const spy = mount(st({ run: run("running"), watched: {} })); await waitFor(() => expect(screen.getByTestId("takeout-running").textContent).toBe("settings.takeout.running"), ); - expect(zone()).toBeNull(); - spy.mockResolvedValue({ run: run("ok"), watched: { watched: 5 } }); + expect(unavailable(zone())).toBe(true); + spy.mockResolvedValue(st({ run: run("ok"), watched: { watched: 5 } })); await waitFor(() => expect(line()).toContain("settings.takeout.count.watched(5,5)"), { timeout: 3000, }); @@ -143,7 +155,7 @@ describe("YouTube history import · what it says", () => { }); it("a re-read that fails does not leave the previous answer standing under it", async () => { - const status = mount({ run: run("ok", { finished_at: "OLD" }), watched: {} }); + const status = mount(st({ run: run("ok", { finished_at: "OLD" }), watched: {} })); await waitFor(() => expect(line()).toContain("OLD")); fireEvent.click(screen.getByRole("button", { name: "settings.takeout.newImport" })); vi.spyOn(api, "takeoutUpload").mockResolvedValue(uploaded(1)); @@ -158,7 +170,7 @@ describe("YouTube history import · what it says", () => { describe("YouTube history import · the upload", () => { it("sends each dropped file on its own, says each outcome, and reads the new run", async () => { - const status = mount({ run: null, watched: {} }); + const status = mount(st({ run: null, watched: {} })); await waitFor(() => expect(zone()).toBeTruthy()); const refused = new HttpError(422, "refused"); const sent: string[] = []; @@ -167,7 +179,7 @@ describe("YouTube history import · the upload", () => { if (f.name === "bad.html") throw refused; return uploaded(f.name === "watch-history.json" ? 9034 : 56); }); - status.mockResolvedValue({ run: run("ok"), watched: { watched: 7 } }); + status.mockResolvedValue(st({ run: run("ok"), watched: { watched: 7 } })); await drop([file("watch-history.json"), file("bad.html"), file("Watch later-videos.csv")]); await waitFor(() => expect(line()).toContain("settings.takeout.count.watched(7,7)")); @@ -183,7 +195,7 @@ describe("YouTube history import · the upload", () => { }); it("names the file being sent, then never shows the previous import's line before the new one", async () => { - const status = mount({ run: run("ok", { finished_at: "OLD" }), watched: {} }); + const status = mount(st({ run: run("ok", { finished_at: "OLD" }), watched: {} })); await waitFor(() => expect(line()).toContain("OLD")); fireEvent.click(screen.getByRole("button", { name: "settings.takeout.newImport" })); let finish!: (v: TakeoutUpload) => void; @@ -197,7 +209,115 @@ describe("YouTube history import · the upload", () => { await act(async () => finish(uploaded(1))); // The new run is still being read: the upload's line stays, the old import's does not return. expect(screen.queryByTestId("takeout-last")).toBeNull(); - await act(async () => answer({ run: run("ok", { finished_at: "NEW" }), watched: {} })); + await act(async () => answer(st({ run: run("ok", { finished_at: "NEW" }), watched: {} }))); await waitFor(() => expect(line()).toContain("NEW")); }); }); + +describe("YouTube history import · while it runs", () => { + it("a queued import is this upload's, not the last one: the running line until its run reads", async () => { + const spy = mount(st({ queued: true, run: run("ok", { finished_at: "OLD" }), watched: {} })); + await waitFor(() => + expect(screen.getByTestId("takeout-running").textContent).toBe("settings.takeout.running"), + ); + expect(screen.queryByTestId("takeout-last")).toBeNull(); + spy.mockResolvedValue(st({ run: run("ok", { finished_at: "NEW" }), watched: {} })); + await waitFor(() => expect(line()).toContain("NEW"), { timeout: 3000 }); + }); + + it("a re-read that fails mid-run does not leave the spinner claiming progress under it", async () => { + const spy = mount(st({ run: run("running"), watched: {} })); + await waitFor(() => expect(screen.getByTestId("takeout-running")).toBeTruthy()); + spy.mockRejectedValue(new HttpError(500, "boom")); + await waitFor( + () => expect(screen.getByRole("alert").textContent).toContain("settings.takeout.loadFailed"), + { timeout: 3000 }, + ); + expect(screen.queryByTestId("takeout-running")).toBeNull(); + spy.mockResolvedValue(st({ run: run("ok"), watched: {} })); + await waitFor(() => expect(screen.queryByRole("alert")).toBeNull(), { timeout: 3000 }); + }); + + it("one live region says the run and then its outcome", async () => { + const spy = mount(st({ run: run("running"), watched: {} })); + await waitFor(() => expect(screen.getByTestId("takeout-running")).toBeTruthy()); + const region = screen.getByRole("status"); + spy.mockResolvedValue(st({ run: run("ok"), watched: { watched: 2 } })); + await waitFor(() => expect(line()).toContain("watched(2,2)"), { timeout: 3000 }); + expect(screen.getByRole("status")).toBe(region); + expect(region.contains(screen.getByTestId("takeout-last"))).toBe(true); + }); + + it("a file dropped beside the zone is kept from the browser, which would open it", async () => { + mount(st({ run: null, watched: {} })); + await waitFor(() => expect(zone()).toBeTruthy()); + const missed = (type: string, types: string[]) => { + const e = new Event(type, { bubbles: true, cancelable: true }); + Object.defineProperty(e, "dataTransfer", { value: { types, dropEffect: "copy" } }); + document.body.dispatchEvent(e); + return e; + }; + expect(missed("drop", ["Files"]).defaultPrevented).toBe(true); + expect(missed("dragover", ["Files"]).defaultPrevented).toBe(true); + // A drag of something else (a link, text) is the page's own business. + expect(missed("drop", ["text/plain"]).defaultPrevented).toBe(false); + cleanup(); + expect(missed("drop", ["Files"]).defaultPrevented).toBe(false); + }); + + it("a file larger than the import takes is refused before it is sent, naming the limit", async () => { + mount(st({ max_bytes: 10, run: run("ok"), watched: {} })); + await waitFor(() => expect(line()).toBeTruthy()); + fireEvent.click(screen.getByRole("button", { name: "settings.takeout.newImport" })); + const send = vi.spyOn(api, "takeoutUpload").mockResolvedValue(uploaded(1)); + await drop([file("watch-history.json", 11)]); + expect(send).not.toHaveBeenCalled(); + expect(mocks.notify).toHaveBeenCalledWith({ + level: "error", + message: "settings.takeout.tooLarge(watch-history.json,10 B)", + }); + // Nothing was taken: the zone stays open for the right file, and so after a refusal. + await waitFor(() => expect(zone()).toBeTruthy()); + send.mockRejectedValue(new HttpError(422, "refused")); + await drop([file("bad.html")]); + await waitFor(() => expect(mocks.notifyActionError).toHaveBeenCalled()); + expect(zone()).toBeTruthy(); + }); + + it("the focus follows: New import hands it to the zone, the zone back to New import", async () => { + const spy = mount(st({ run: run("ok", { finished_at: "OLD" }), watched: {} })); + await waitFor(() => expect(line()).toContain("OLD")); + fireEvent.click(screen.getByRole("button", { name: "settings.takeout.newImport" })); + expect(document.activeElement).toBe(zone()); + + vi.spyOn(api, "takeoutUpload").mockResolvedValue(uploaded(1)); + spy.mockResolvedValue( + st({ queued: true, run: run("ok", { finished_at: "OLD" }), watched: {} }), + ); + await drop([file("watch-history.json")]); + // While it runs the zone stays, unavailable, and keeps the focus. + expect(document.activeElement).toBe(zone()); + expect(unavailable(zone())).toBe(true); + spy.mockResolvedValue(st({ run: run("ok", { finished_at: "NEW" }), watched: {} })); + await waitFor(() => expect(line()).toContain("NEW"), { timeout: 3000 }); + expect(zone()).toBeNull(); + expect(document.activeElement).toBe( + screen.getByRole("button", { name: "settings.takeout.newImport" }), + ); + }); + + it("a focus the user moved elsewhere meanwhile is not taken back", async () => { + const spy = mount(st({ run: null, watched: {} })); + await waitFor(() => expect(zone()).toBeTruthy()); + const elsewhere = document.body.appendChild(document.createElement("button")); + zone()!.focus(); + vi.spyOn(api, "takeoutUpload").mockResolvedValue(uploaded(1)); + spy.mockResolvedValue(st({ queued: true, run: null, watched: {} })); + await drop([file("watch-history.json")]); + elsewhere.focus(); + spy.mockResolvedValue(st({ run: run("ok", { finished_at: "NEW" }), watched: {} })); + await waitFor(() => expect(line()).toContain("NEW"), { timeout: 3000 }); + expect(document.activeElement).toBe(elsewhere); + elsewhere.remove(); + }); +}); diff --git a/frontend/src/modules/settings/TakeoutImport.tsx b/frontend/src/modules/settings/TakeoutImport.tsx index 2c57466a..9bab8821 100644 --- a/frontend/src/modules/settings/TakeoutImport.tsx +++ b/frontend/src/modules/settings/TakeoutImport.tsx @@ -1,4 +1,4 @@ -import { useEffect, useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { useTranslation } from "react-i18next"; import { useQuery, useQueryClient } from "@tanstack/react-query"; import { RefreshCw, UploadCloud } from "lucide-react"; @@ -6,19 +6,19 @@ import { Section } from "../../components/ui/form"; import { Button } from "../../components/ui/Button"; import { DropZone } from "../../components/ui/DropZone"; import { StateMessage } from "../../components/ui/QueryState"; -import { api, type TakeoutStatus } from "../../lib/api"; +import { api, type TakeoutStatus, type TakeoutUpload } from "../../lib/api"; +import type { TAKEOUT_OUTCOMES } from "../../lib/api/feed"; import { qk } from "../../lib/queryKeys"; import { notify } from "../../lib/notifications"; import { notifyActionError } from "../../lib/actionErrors"; -import { formatDateTime, formatNumber } from "../../lib/format"; +import { formatBytes, formatDateTime, formatNumber } from "../../lib/format"; +import { importBusy, invalidateImported, watchTakeout } from "./takeoutWatch"; // Settings › Account: the user's Google Takeout export, imported. What they watched on YouTube and // their YouTube Watch later list reach no API, so the export's two files are uploaded here, one // request each; the server tells them apart and runs the import in the background, and its outcome // also reaches the inbox. The import has no steps to count (the whole history is applied in one -// go), so while it runs the section says so and asks again until it is done. - -const POLL_MS = 1500; +// go), so while it runs the section says so and asks again until it is done (`takeoutWatch`). // The outcome counts, in the order they are said: the history's, then Watch later's. A count the // import did not produce is not said at all. @@ -35,6 +35,12 @@ const WATCH_LATER = [ ["waiting", "laterWaiting"], ] as const; +// Every outcome the server can name is said by one of the two lists (feed.ts holds them to it). +type TakeoutOutcome = (typeof TAKEOUT_OUTCOMES)[number]; +type Said = (typeof HISTORY)[number][0] | (typeof WATCH_LATER)[number][0]; +const everyOutcomeSaid: [Exclude] extends [never] ? true : never = true; +void everyOutcomeSaid; + function countParts(s: TakeoutStatus, say: (key: string, n: number) => string): string[] { const parts: string[] = []; for (const [outcome, key] of HISTORY) { @@ -52,58 +58,111 @@ export default function TakeoutImport() { const { t, i18n } = useTranslation(); const lang = i18n.language; const qc = useQueryClient(); - const status = useQuery({ - queryKey: qk.takeout(), - queryFn: api.takeoutStatus, - refetchInterval: (q) => (q.state.data?.run?.status === "running" ? POLL_MS : false), - }); + const status = useQuery({ queryKey: qk.takeout(), queryFn: api.takeoutStatus }); const [uploading, setUploading] = useState(null); // "New import" reopens the drop zone over a finished one; before the first import it is open. const [reopened, setReopened] = useState(false); const run = status.data?.run ?? null; - const running = uploading !== null || run?.status === "running"; + // Queued counts too: until its run row is written, `run` is still the import before it. + const busy = importBusy(status.data); + const running = uploading !== null || busy; const finishedAt = run?.finished_at ?? null; - // A run that ended has changed what the feed and Watch later show: those lists are read again - // when the user next opens them, not 30 s later from the cache. + // A run under way is watched to its end, even if the user leaves Settings meanwhile. useEffect(() => { - if (!finishedAt) return; - void qc.invalidateQueries({ queryKey: qk.feed() }); - void qc.invalidateQueries({ queryKey: qk.playlist() }); + if (busy) void watchTakeout(qc); + }, [busy, qc]); + // One that ended while nothing here watched it (a reload, another tab) has changed what the feed + // and Watch later show too: those lists are read again when next opened, not from the cache. + useEffect(() => { + if (finishedAt) invalidateImported(qc); }, [finishedAt, qc]); + // A file dropped beside the zone, or while it is not there, is kept from the browser: it would + // open the file in the tab, unloading the app and the upload under way. The zone takes its own. + useEffect(() => { + const missed = (e: DragEvent) => { + if (e.defaultPrevented || !e.dataTransfer?.types.includes("Files")) return; + e.preventDefault(); + if (e.type === "dragover") e.dataTransfer.dropEffect = "none"; + }; + window.addEventListener("dragover", missed); + window.addEventListener("drop", missed); + return () => { + window.removeEventListener("dragover", missed); + window.removeEventListener("drop", missed); + }; + }, []); + + // Where the focus goes when what holds it leaves: "New import" hands it to the zone it opens, and + // the zone, gone once the import it started is done, hands it to "New import". + const zone = useRef(null); + const focusNext = useRef<"zone" | "newImport" | null>(null); + const focusIfNext = (which: "zone" | "newImport", el: HTMLButtonElement | null) => { + if (!el || focusNext.current !== which) return; + focusNext.current = null; + // Only focus that was lost with the control that held it: never taken from where the user went. + if (document.activeElement === null || document.activeElement === document.body) el.focus(); + }; + const zoneRef = (el: HTMLButtonElement | null) => { + zone.current = el; + focusIfNext("zone", el); + }; + const newImportRef = (el: HTMLButtonElement | null) => focusIfNext("newImport", el); + const upload = async (files: File[]) => { - for (const file of files) await uploadOne(file); + if (document.activeElement === zone.current) focusNext.current = "newImport"; + let taken = false; + for (const file of files) if (await uploadOne(file)) taken = true; // The new run read first, then the upload's line let go: the other way round the previous // import's line showed for a moment, as if this upload had done nothing. await qc.invalidateQueries({ queryKey: qk.takeout() }); setUploading(null); - setReopened(false); + // Nothing taken, nothing imported: the zone stays open for the right file. + if (taken) setReopened(false); + else focusNext.current = null; }; - const uploadOne = async (file: File) => { - setUploading(file.name); - try { - const res = await api.takeoutUpload(file); + /** Send one file; whether the server took it. */ + const uploadOne = async (file: File): Promise => { + const max = status.data?.max_bytes ?? Infinity; + if (file.size > max) { + // Refused here, not after minutes of upload, and with the limit named. notify({ - level: "success", - message: t("settings.takeout.read", { - name: file.name, - count: res.videos, - n: formatNumber(res.videos, lang), - }), + level: "error", + message: t("settings.takeout.tooLarge", { name: file.name, limit: formatBytes(max, lang) }), }); + return false; + } + setUploading(file.name); + let res: TakeoutUpload | null = null; + try { + res = await api.takeoutUpload(file); } catch (e) { // A refusal names its reason (not a Takeout file, nothing in it); the next file still goes. notifyActionError(e, t("errors.writeFailed.takeoutUpload")); } + if (res === null) return false; + notify({ + level: "success", + message: t("settings.takeout.read", { + name: file.name, + count: res.videos, + n: formatNumber(res.videos, lang), + }), + }); + return true; }; const say = (key: string, n: number) => t(`settings.takeout.count.${key}`, { count: n, n: formatNumber(n, lang) }); const when = (iso: string | null) => (iso ? formatDateTime(iso, lang) : ""); // Not while the last import is still being asked for: a zone that flashes before the line that - // replaces it would offer a first import to someone who has done one. - const showZone = status.isSuccess && !running && (run === null || reopened); + // replaces it would offer a first import to someone who has done one. While an import runs the + // zone stays, unavailable: the control that started it keeps the focus. + const showZone = status.isSuccess && (running || run === null || reopened); + // A failed re-read keeps the previous answer: it is not shown under the line saying the read + // failed, as if it were current — a run's spinner no more than its outcome. + const current = !status.isError || uploading !== null; return (
@@ -117,7 +176,14 @@ export default function TakeoutImport() { {showZone && ( <> - + {(dragOver) => ( <> @@ -138,42 +204,49 @@ export default function TakeoutImport() { )} - {running ? ( -
- - {uploading !== null - ? t("settings.takeout.uploading", { name: uploading }) - : t("settings.takeout.running")} +
+ {/* One live region for the run and its outcome, mounted throughout: what replaces + "Processing…" is announced, which a region mounted with its text is not. */} +
+ {running && current ? ( + + + {uploading !== null + ? t("settings.takeout.uploading", { name: uploading }) + : t("settings.takeout.running")} + + ) : ( + !running && + run && + status.data && + current && ( +

+ {run.status === "error" + ? t("settings.takeout.failed", { when: when(run.finished_at) }) + : run.status === "interrupted" + ? t("settings.takeout.interrupted", { when: when(run.started_at) }) + : [ + t("settings.takeout.last", { when: when(run.finished_at) }), + ...countParts(status.data, say), + ].join(" · ")} +

+ ) + )}
- ) : ( - // A failed re-read keeps the previous answer: it is not shown under the line saying the - // read failed, as if it were current. - run && - status.data && - !status.isError && ( -
-

- {run.status === "error" - ? t("settings.takeout.failed", { when: when(run.finished_at) }) - : run.status === "interrupted" - ? t("settings.takeout.interrupted", { when: when(run.started_at) }) - : [ - t("settings.takeout.last", { when: when(run.finished_at) }), - ...countParts(status.data, say), - ].join(" · ")} -

- {!reopened && ( - - )} -
- ) - )} + {!running && run && status.data && current && !reopened && ( + + )} +
); } diff --git a/frontend/src/modules/settings/takeoutWatch.test.ts b/frontend/src/modules/settings/takeoutWatch.test.ts new file mode 100644 index 00000000..c2d435d2 --- /dev/null +++ b/frontend/src/modules/settings/takeoutWatch.test.ts @@ -0,0 +1,67 @@ +// @vitest-environment jsdom +// +// A Takeout import is watched to its end whether or not its section is still on screen, and what it +// changed is read afresh then: the feed and its counts, Watch later, and the playlists' sizes. +import { afterEach, describe, expect, it, vi } from "vitest"; +import { QueryClient } from "@tanstack/react-query"; +import { api, type TakeoutStatus } from "../../lib/api"; +import { qk } from "../../lib/queryKeys"; +import { importBusy, watchTakeout } from "./takeoutWatch"; + +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); + +const st = (s: Partial): TakeoutStatus => ({ + queued: false, + max_bytes: 1, + run: null, + ...s, +}); +const running = { + status: "running", + started_at: "S", + finished_at: null, + result: null, + failure: null, +}; + +describe("the Takeout import's watch", () => { + it("a queued import is busy as much as a running one", () => { + expect(importBusy(st({ queued: true }))).toBe(true); + expect(importBusy(st({ run: { ...running, status: "running" } as TakeoutStatus["run"] }))).toBe( + true, + ); + expect(importBusy(st({ run: { ...running, status: "ok" } as TakeoutStatus["run"] }))).toBe( + false, + ); + expect(importBusy(undefined)).toBe(false); + }); + + it("asks until the import is done, then reads again everything it changed", async () => { + vi.useFakeTimers(); + const qc = new QueryClient(); + qc.setQueryData(qk.takeout(), st({ queued: true })); + const answers = [st({ run: running as TakeoutStatus["run"] }), st({})]; + const read = vi.spyOn(api, "takeoutStatus").mockImplementation(async () => answers.shift()!); + const stale = vi.spyOn(qc, "invalidateQueries"); + + const done = watchTakeout(qc); + void watchTakeout(qc); // a second call while it watches is the same watch + await vi.advanceTimersByTimeAsync(1500); + expect(read).toHaveBeenCalledTimes(1); + expect(stale).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1500); + await done; + + expect(read).toHaveBeenCalledTimes(2); + expect(stale.mock.calls.map(([f]) => f?.queryKey)).toEqual([ + qk.feed(), + qk.feedSummary(), + qk.feedYears(), + qk.playlist(), + qk.playlists(), + ]); + }); +}); diff --git a/frontend/src/modules/settings/takeoutWatch.ts b/frontend/src/modules/settings/takeoutWatch.ts new file mode 100644 index 00000000..13855b2f --- /dev/null +++ b/frontend/src/modules/settings/takeoutWatch.ts @@ -0,0 +1,57 @@ +import type { QueryClient } from "@tanstack/react-query"; +import { api, type TakeoutStatus } from "../../lib/api"; +import { qk } from "../../lib/queryKeys"; + +// A Takeout import, watched to its end: its status is asked again until no run is queued or going, +// then the lists it changed are read afresh. It lives outside the section on purpose: the user may +// leave Settings while the import runs, and the feed they go to must not keep showing as new what +// the import has just marked watched. + +const POLL_MS = 1500; + +export const importBusy = (s: TakeoutStatus | undefined): boolean => + !!s && (s.queued || s.run?.status === "running"); + +let watching = false; + +const pause = () => new Promise((done) => setTimeout(done, POLL_MS)); + +async function read(qc: QueryClient): Promise { + try { + return await qc.fetchQuery({ + queryKey: qk.takeout(), + queryFn: api.takeoutStatus, + staleTime: 0, + }); + } catch { + // A read that failed is the section's to say (its query holds the error); the watch asks again. + return qc.getQueryData(qk.takeout()); + } +} + +/** Mark stale everything an import may have changed: the feed and its counts, Watch later, and the + * playlists that show its size. */ +export function invalidateImported(qc: QueryClient): void { + for (const queryKey of [ + qk.feed(), + qk.feedSummary(), + qk.feedYears(), + qk.playlist(), + qk.playlists(), + ]) + void qc.invalidateQueries({ queryKey }); +} + +/** Watch the user's import until it is done (once: a second call while it watches is that watch), + * then mark stale what it changed. */ +export async function watchTakeout(qc: QueryClient): Promise { + if (watching) return; + watching = true; + let status = qc.getQueryData(qk.takeout()); + while (importBusy(status)) { + await pause(); + status = await read(qc); + } + watching = false; + invalidateImported(qc); +} diff --git a/scripts/e2e_seed.py b/scripts/e2e_seed.py index c4a1be6c..8fa5236c 100644 --- a/scripts/e2e_seed.py +++ b/scripts/e2e_seed.py @@ -181,6 +181,28 @@ def _seed_account(db, email: str, role: str, display_name: str) -> dict: saved = svid.id db.commit() + # 5. Videos the Takeout import marks watched: from the subscribed channels, and already cleared by + # the Shorts probe — one not yet probed would wait for it, and the import spec would count + # nothing. One per browser project: they share this account, and a video imported once is not + # marked again. + known = ( + list( + db.scalars( + select(Video.id) + .where( + Video.channel_id.in_(channel_ids), + Video.shorts_probed.is_(True), + Video.is_short.is_(False), + Video.id.not_in(used | {saved} if saved else used), + ) + .order_by(Video.published_at.desc()) + .limit(5) + ) + ) + if channel_ids + else [] + ) + return { "user_id": user.id, "email": email, @@ -188,6 +210,7 @@ def _seed_account(db, email: str, role: str, display_name: str) -> dict: "in_progress": in_progress, "watched": watched, "saved": saved, + "takeout_known": known, }