"""Reader's writes: a user's labels and saved views, where they are in a video, their playlists' items and order — and the system tags the catalog's auto-tagger hands over. Every function here flushes and leaves the commit to its caller (the handler, or the job that owns its session). A write that must not double up keeps its guard where it is written: an upsert on the unique key, a row lock taken before a presence check. The comfort shelf's writes are Home's (`discover.pins`); a playlist's YouTube sync is `sync.playlists`. """ from datetime import datetime, timezone from typing import cast from sqlalchemy import Table, bindparam, delete, exists, func, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app.models import User from app.reader import queries from app.reader.models import ( ChannelTag, Playlist, PlaylistItem, SavedView, Tag, TakeoutChannelDismissal, TakeoutItem, TakeoutKind, TakeoutOutcome, VideoState, VideoUnmark, ) from app.sync.playlists import lock_playlists, recompute_dirty from app.userscope import owned_by TAG_CATEGORIES = ("language", "topic", "other") VIEW_NAME_MAX = 80 # --- tags ------------------------------------------------------------------------------------ def create_tag(db: Session, user: User, name: str, category: str, color: str | None) -> Tag: """A tag of the user's own. An unknown category falls back to "other" rather than refusing. A name the user already has raises `IntegrityError` at the flush.""" tag = Tag( user_id=user.id, name=name, color=color, category=category if category in TAG_CATEGORIES else "other", ) db.add(tag) db.flush() return tag def update_tag(tag: Tag, *, sent: set[str], name: str | None, color: str | None) -> None: """Change the fields the caller sent: `color` sent as None clears it.""" if "name" in sent and name is not None: tag.name = name if "color" in sent: tag.color = color def delete_tag(db: Session, tag: Tag) -> None: db.delete(tag) # its channel links cascade db.flush() def ensure_system_tag(db: Session, name: str, category: str, color: str | None) -> Tag: """The shared system tag of this name, made the first time a channel earns it.""" tag = db.execute( select(Tag).where(Tag.user_id.is_(None), Tag.name == name) ).scalar_one_or_none() if tag is None: tag = Tag(user_id=None, name=name, category=category, color=color) db.add(tag) db.flush() return tag def set_auto_tags(db: Session, channel_id: str, desired: dict[int, float | None]) -> None: """Make a channel's automatic tag links exactly `desired` (tag id → confidence). Only the auto-tagger's own links move; a user's links on the channel are never touched.""" existing = { ct.tag_id: ct for ct in db.execute( select(ChannelTag).where( ChannelTag.channel_id == channel_id, ChannelTag.user_id.is_(None), ChannelTag.source == "auto", ) ).scalars() } for tag_id, ct in existing.items(): if tag_id not in desired: db.delete(ct) for tag_id, conf in desired.items(): ct = existing.get(tag_id) if ct is None: db.add( ChannelTag( channel_id=channel_id, tag_id=tag_id, user_id=None, source="auto", confidence=conf, ) ) else: ct.confidence = conf db.flush() def prune_orphan_system_tags(db: Session) -> int: """Delete the system tags no channel links to any more (after a mapping change) — but not one a saved feed view filters on. Views hold tag IDS, and a deleted tag comes back under a new id when a channel earns it again, so the view's filter would point at nothing; the feed drops an unknown id, and the view showed the whole feed under its old name. The kept ids are filtered here rather than in SQL: a `NOT IN` over every view's ids could pass the driver's bind-parameter limit.""" in_views = queries.tags_in_feed_views(db) orphans = [ tag for tag in db.execute( select(Tag).where(Tag.user_id.is_(None), ~exists().where(ChannelTag.tag_id == Tag.id)) ).scalars() if tag.id not in in_views ] for tag in orphans: db.delete(tag) db.flush() return len(orphans) # --- saved views ----------------------------------------------------------------------------- def create_view(db: Session, user: User, source: str, name: str, filters: dict) -> SavedView: """A view appended after the user's others of the same module (each module numbers its own 0..N). A name the user already has there raises `IntegrityError` at the flush.""" maxpos = db.execute( select(func.coalesce(func.max(SavedView.position), -1)).where( owned_by(SavedView.user_id, user), SavedView.source == source ) ).scalar_one() view = SavedView( user_id=user.id, source=source, name=name[:VIEW_NAME_MAX], filters=filters, position=maxpos + 1, ) db.add(view) db.flush() return view def update_view( db: Session, view: SavedView, *, sent: set[str], name: str | None = None, filters: dict | None = None, is_default: bool | None = None, ) -> None: """Change the fields the caller sent. Making a view the default clears the flag on the user's other views of the same module first: at most one default per (user, source).""" if "name" in sent and name is not None: view.name = name[:VIEW_NAME_MAX] if "filters" in sent and filters is not None: view.filters = filters if "is_default" in sent: make_default = bool(is_default) if make_default: db.execute( update(SavedView) .where( owned_by(SavedView.user_id, view.user_id), SavedView.source == view.source, SavedView.id != view.id, ) .values(is_default=False) ) view.is_default = make_default db.flush() def delete_view(db: Session, view: SavedView) -> None: db.delete(view) db.flush() def reorder_views(db: Session, user: User, ids: list[int]) -> None: """Number the listed views 0..n in the listed order; ids that are not the user's are skipped.""" by_id = {v.id: v for v in queries.all_views(db, user)} pos = 0 for vid in ids: v = by_id.get(vid) if v is not None: v.position = pos pos += 1 db.flush() # --- watch state ----------------------------------------------------------------------------- def unmark(db: Session, user: User, video_id: str) -> None: """Back to "new". A row that still holds a resume point keeps it (un-marking "watched" restores the in-progress state, not where the user left off); otherwise the row goes. The player fires this beside a checkpoint that may delete the same row: the caller's commit then raises `StaleDataError`, which is that lost race and nothing worse.""" lock_takeout(db, user) # an import settling meanwhile reads the un-mark below, or comes first row = queries.video_state(db, user, video_id) if row is None: return if row.position_seconds: row.status = "new" row.watched_at = None else: db.delete(row) _remember_unmark(db, user, video_id) db.flush() def _remember_unmark(db: Session, user: User, video_id: str) -> None: """Note that the user took the video back to "new" now: the state they set leaves no row, and an import of their YouTube history must not mark it watched again (`apply_takeout_watched`).""" now = datetime.now(timezone.utc) db.execute( pg_insert(VideoUnmark) .values(user_id=user.id, video_id=video_id, at=now) .on_conflict_do_update(constraint="uq_video_unmark", set_={"at": now}) ) def mark(db: Session, user: User, video_id: str, status: str) -> None: """ "watched" or "hidden", as one atomic upsert on (user, video). The player marks "watched" at the end of playback at the instant a checkpoint may delete the in-progress row; a SELECT then UPDATE raised `StaleDataError` on that race, and an INSERT … ON CONFLICT is immune to it. "hidden" keeps any earlier `watched_at`.""" now = datetime.now(timezone.utc) set_: dict = {"status": status, "updated_at": now} if status == "watched": set_["watched_at"] = now db.execute( pg_insert(VideoState) .values( user_id=user.id, video_id=video_id, status=status, watched_at=now if status == "watched" else None, ) .on_conflict_do_update(constraint="uq_user_video", set_=set_) ) def clear_state(db: Session, user: User, video_id: str) -> None: """The video back to pristine for this user: status, resume point and `watched_at` all go. Playlist membership lives elsewhere and stays.""" lock_takeout(db, user) # as `unmark` row = queries.video_state(db, user, video_id) if row is not None: db.delete(row) _remember_unmark(db, user, video_id) db.flush() def set_progress(db: Session, user: User, video_id: str, position: int) -> None: """The player's resume point. 0 clears it: a row holding nothing else goes, one with a status (watched, hidden) keeps it with the position zeroed. The status itself is never changed here. A position is one upsert on (user, video): the first checkpoint and the end-of-playback mark beside it both read "no row", and a plain add let the second one fail on `uq_user_video`.""" now = datetime.now(timezone.utc) if position: db.execute( pg_insert(VideoState) .values( user_id=user.id, video_id=video_id, position_seconds=position, progress_updated_at=now, ) .on_conflict_do_update( constraint="uq_user_video", set_={"position_seconds": position, "progress_updated_at": now, "updated_at": now}, ) ) return row = queries.video_state(db, user, video_id) if row is None: return if row.status == "new": db.delete(row) else: row.position_seconds = 0 row.progress_updated_at = now db.flush() # --- playlists ------------------------------------------------------------------------------- class MoveRefused(Exception): """A move between playlists that cannot happen: `reason` is the code the UI translates.""" def __init__(self, reason: str): super().__init__(reason) self.reason = reason def touch(pl: Playlist) -> None: """Bump the playlist's `updated_at` for an edit that only writes `playlist_items`. `UpdatedAtMixin`'s `onupdate` fires on an UPDATE of the PLAYLIST row, and add / remove / reorder / move write the item rows instead — so without this the rail's "recently changed" sort would only ever see renames. The assignment is what makes the column mean "last time this list changed", which is what the rail promises the reader. The DATABASE's clock, like the mixin's own `onupdate`: one sort key written from two clocks (this process's and Postgres's) could order a rename and an item edit the wrong way round.""" pl.updated_at = func.now() def _next_playlist_position(db: Session, user_id: int) -> int: """Append position for a new playlist in this user's rail.""" return ( db.execute( select(func.coalesce(func.max(Playlist.position), -1)).where( owned_by(Playlist.user_id, user_id) ) ).scalar_one() + 1 ) def next_item_position(db: Session, playlist_id: int) -> int: """Append position for a new item at the end of a playlist. Locks the playlist row first: two adds (two tabs, the rail and the page) otherwise both read the same max and land on one position, and the list's order between them is then arbitrary.""" lock_playlists(db, playlist_id) return ( db.execute( select(func.coalesce(func.max(PlaylistItem.position), -1)).where( PlaylistItem.playlist_id == playlist_id ) ).scalar_one() + 1 ) def create_playlist(db: Session, user: User, name: str) -> Playlist: pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id)) db.add(pl) db.flush() return pl def watch_later(db: Session, user: User) -> Playlist: """The user's built-in Watch later playlist, created on first use. Created by the insert itself (ON CONFLICT DO NOTHING on the one-per-user index) and then read: the first two saves at once both read "none", and each used to make a list of its own.""" pl = queries.watch_later(db, user) if pl is None: db.execute( pg_insert(Playlist) .values( user_id=user.id, name="Watch later", kind="watch_later", position=_next_playlist_position(db, user.id), ) .on_conflict_do_nothing( index_elements=["user_id"], index_where=text("kind = 'watch_later'") ) ) pl = queries.watch_later(db, user) assert pl is not None # inserted now, or committed by the request that won return pl def add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool: """Append a video to a playlist if not already present (idempotent). Returns whether it was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later). Locked BEFORE the presence check: two adds of one video both saw "absent" and the second one's insert hit uq_playlist_video as a 500.""" lock_playlists(db, pl.id) if queries.item(db, pl.id, video_id) is not None: return False db.add( PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_item_position(db, pl.id)) ) if mark_dirty: recompute_dirty(db, pl) touch(pl) db.flush() return True def remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool: """Remove a video from a playlist if present (idempotent). Returns whether it was removed. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).""" lock_playlists(db, pl.id) found = queries.item(db, pl.id, video_id) if found is None: return False db.delete(found) if mark_dirty: recompute_dirty(db, pl) touch(pl) db.flush() return True def place_item(db: Session, pl: Playlist, video_id: str, position: int) -> None: """Put a video already in the list at `position` (clamped to the end), renumbering the whole list: positions can have gaps after removals, so "index N" is only meaningful over the ordered rows.""" rows = queries.items(db, pl.id) new = next(r for r in rows if r.video_id == video_id) rows.remove(new) rows.insert(min(position, len(rows)), new) for pos, row in enumerate(rows): row.position = pos db.flush() def add_to_playlist(db: Session, pl: Playlist, video_id: str, position: int | None) -> None: """Add a video to one of the user's lists, at `position` when given (the list page's "Undo" of a removal puts the row back where it was in ONE request), else at the end. A video already in the list stays where it is.""" added = add_item(db, pl, video_id, mark_dirty=False) if added and position is not None: place_item(db, pl, video_id, position) recompute_dirty(db, pl) db.flush() def rename_playlist(db: Session, pl: Playlist, name: str) -> None: """A new name moves `dirty`; the caller holds the playlist's lock (the scheduled mirror refresh reads `dirty` under it).""" if name != pl.name: pl.name = name recompute_dirty(db, pl) db.flush() def delete_playlist(db: Session, pl: Playlist) -> None: """The playlist first, as every item write: the cascade deletes the items after the list row is locked.""" lock_playlists(db, pl.id) db.delete(pl) # items cascade db.flush() def move_item(db: Session, src: Playlist, dst: Playlist, video_id: str) -> None: """Move one video from `src` into `dst`, in the caller's one transaction. The client-side shape — add to the target, then remove from the source — leaves the video in BOTH lists whenever the second call fails. Idempotent: a video already in the target is simply dropped from the source (moving into a list that has it is a removal, not a duplicate). Raises `MoveRefused` for the same list twice, or a video a concurrent edit already took out.""" if src.id == dst.id: raise MoveRefused("same_playlist") # Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise. lock_playlists(db, src.id, dst.id) row = queries.item(db, src.id, video_id) if row is None: raise MoveRefused("not_in_playlist") dst_changed = False if queries.item(db, dst.id, video_id) is None: # Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this") # survives the move, and there is no window where the video belongs to neither list. The # check above holds: both lists are locked, and an insert into the target (another tab's # add, a YouTube merge) waits for the lock — its foreign key needs the target row. row.playlist_id = dst.id row.position = next_item_position(db, dst.id) dst_changed = True if not dst_changed: db.delete(row) # The source always changed; the target only when the row landed in it. Touching a target # that already had the video promoted an untouched list in the rail's "recently changed" sort. recompute_dirty(db, src) touch(src) if dst_changed: recompute_dirty(db, dst) touch(dst) db.flush() def reorder_items(db: Session, pl: Playlist, order: list[str]) -> int: """Number the listed videos 0..n in the listed order. Items the list left out (added in another tab meanwhile) keep their relative order after them, so no two items share a position. The caller holds the playlist's lock. Returns how many items the list has.""" by_video = {it.video_id: it for it in queries.items(db, pl.id)} pos = 0 seen: set[str] = set() for vid in order: it = by_video.get(vid) if it is not None: it.position = pos pos += 1 seen.add(vid) for it in sorted( (it for vid, it in by_video.items() if vid not in seen), key=lambda i: i.position ): it.position = pos pos += 1 recompute_dirty(db, pl) touch(pl) db.flush() return pos def attach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None: """The user's own label on a channel (idempotent). The caller has checked the tag is theirs.""" if queries.user_tag_link(db, user, channel_id, tag_id) is None: db.add(ChannelTag(channel_id=channel_id, tag_id=tag_id, user_id=user.id, source="user")) db.flush() def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None: row = queries.user_tag_link(db, user, channel_id, tag_id) if row is not None: db.delete(row) db.flush() # --- 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. An un-mark takes it too: the import reads the un-marks before it marks, and one landing between the two would be overwritten.""" db.execute(select(func.pg_advisory_xact_lock(_TAKEOUT_LOCK_NS, user.id))) def stage_takeout( db: Session, user: User, kind: TakeoutKind, rows: dict[str, tuple[datetime, str | None]] ) -> None: """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() ] for start in range(0, len(values), _TAKEOUT_BATCH): stmt = pg_insert(TakeoutItem).values(values[start : start + _TAKEOUT_BATCH]) db.execute( stmt.on_conflict_do_update( constraint="uq_takeout_item", set_={ "at": func.greatest(TakeoutItem.at, stmt.excluded.at), "channel_id": func.coalesce(stmt.excluded.channel_id, TakeoutItem.channel_id), }, ) ) db.flush() def _settle(db: Session, ids: list[int], outcome: TakeoutOutcome) -> None: for start in range(0, len(ids), _TAKEOUT_BATCH): db.execute( update(TakeoutItem) .where(TakeoutItem.id.in_(ids[start : start + _TAKEOUT_BATCH])) .values(outcome=outcome) ) def apply_takeout_watched(db: Session, user: User) -> dict[str, int]: """Mark watched what the user watched on YouTube, for the videos the catalog knows and has probed for Shorts; the rest wait. `watched_at` is the Takeout time. A Short is not marked (swiping one past is not watching it), and a state the user set here wins: watched or hidden, anything changed after the Takeout time, or an un-mark since then (`VideoUnmark`). 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]] = {} ready = queries.takeout_watched_ready(db, user) unmarked = queries.unmarked_at(db, user, [video_id for _, video_id, _, _ in ready]) kept: list[int] = [] for row_id, video_id, at, is_short in ready: if is_short: short.append(row_id) elif video_id in unmarked and unmarked[video_id] >= at: # Un-marked here since they watched it there: theirs, as a state they set is. kept.append(row_id) else: todo[video_id] = (row_id, at) marked: set[str] = set() items = list(todo.items()) now = datetime.now(timezone.utc) for start in range(0, len(items), _TAKEOUT_BATCH): stmt = pg_insert(VideoState).values( [ {"user_id": user.id, "video_id": vid, "status": "watched", "watched_at": at} for vid, (_, at) in items[start : start + _TAKEOUT_BATCH] ] ) marked.update( db.execute( stmt.on_conflict_do_update( constraint="uq_user_video", set_={ "status": "watched", "watched_at": stmt.excluded.watched_at, "updated_at": now, }, where=(VideoState.status == "new") & (VideoState.updated_at < stmt.excluded.watched_at), ).returning(VideoState.video_id) ).scalars() ) kept += [row_id for vid, (row_id, _) in items if vid not in marked] _settle(db, [todo[vid][0] for vid in marked], TakeoutOutcome.WATCHED) _settle(db, kept, TakeoutOutcome.KEPT) _settle(db, short, TakeoutOutcome.SHORT) db.flush() return {"watched": len(marked), "kept": len(kept), "shorts": len(short)} 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 {} lock_takeout(db, user) ids = [vid for _, vid in rows] known = queries.known_video_ids(db, ids) pl = watch_later(db, user) # What add_item does a video at a time, for the whole list at once: one presence read and one # insert, not four statements a video (a full list is thousands). Locked before the read, as there. next_position = next_item_position(db, pl.id) held = queries.in_playlist(db, pl.id, [vid for vid in ids if vid in known]) done: dict[TakeoutOutcome, list[int]] = {o: [] for o in TakeoutOutcome} for row_id, vid in rows: if vid not in known: if vid in asked: done[TakeoutOutcome.MISSING].append(row_id) elif vid in held: done[TakeoutOutcome.PRESENT].append(row_id) else: db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=next_position)) next_position += 1 held.add(vid) done[TakeoutOutcome.ADDED].append(row_id) if done[TakeoutOutcome.ADDED]: touch(pl) for outcome, ids in done.items(): _settle(db, ids, outcome) db.flush() return { "watch_later_added": len(done[TakeoutOutcome.ADDED]), "watch_later_present": len(done[TakeoutOutcome.PRESENT]), "watch_later_missing": len(done[TakeoutOutcome.MISSING]), } # --- channels watched but not followed --------------------------------------------------------- def record_takeout_lengths( db: Session, user: User, asked: list[str], lengths: dict[str, int | None] ) -> None: """Keep what YouTube said of the user's watched videos `asked` about: each one's length, or none for a video it no longer has (`lengths` leaves it out). Every asked row is marked asked, so a video YouTube lost is not asked about again.""" if not asked: return lock_takeout(db, user) # these rows are an upload's to update too now = datetime.now(timezone.utc) table = cast(Table, TakeoutItem.__table__) db.execute( table.update() .where( owned_by(TakeoutItem.user_id, user), table.c.kind == TakeoutKind.WATCHED, table.c.video_id == bindparam("vid"), ) .values(looked_up_at=now, duration_seconds=bindparam("seconds")), [{"vid": vid, "seconds": lengths.get(vid)} for vid in asked], ) db.flush() def dismiss_watched_channel(db: Session, user: User, channel_id: str) -> None: """Put a channel out of the user's "watched but not followed" list (again is a no-op).""" db.execute( pg_insert(TakeoutChannelDismissal) .values(user_id=user.id, channel_id=channel_id) .on_conflict_do_nothing(constraint="uq_takeout_channel_dismissal") ) db.flush() def restore_watched_channel(db: Session, user: User, channel_id: str) -> None: """Let a channel the user put out of the list back on it.""" db.execute( delete(TakeoutChannelDismissal).where( owned_by(TakeoutChannelDismissal.user_id, user), TakeoutChannelDismissal.channel_id == channel_id, ) ) db.flush()