diff --git a/backend/alembic/versions/0116_followed_explored_channels.py b/backend/alembic/versions/0116_followed_explored_channels.py new file mode 100644 index 00000000..8d62657a --- /dev/null +++ b/backend/alembic/versions/0116_followed_explored_channels.py @@ -0,0 +1,34 @@ +"""An explored channel someone follows is the catalog's own, and so are its videos + +Revision ID: 0116_followed_explored_channels +Revises: 0115_takeout_channels +Create Date: 2026-10-24 + +A follow pulled in from YouTube (the subscription import) left an explored channel flagged as an +exploration's, and a subscribe landing while an exploration's videos came in left those videos +flagged on a channel that no longer was one. Both are held in the code now; this clears what they +left: the channel of a subscription is not an exploration's, and no video is an exploration's on a +channel that is not one. Data only, so the downgrade has nothing to undo. +""" + +from alembic import op + +revision: str = "0116_followed_explored_channels" +down_revision: str | None = "0115_takeout_channels" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.execute( + "UPDATE channels SET from_explore = false WHERE from_explore" + " AND EXISTS (SELECT 1 FROM subscriptions s WHERE s.channel_id = channels.id)" + ) + op.execute( + "UPDATE videos SET via_explore = false WHERE via_explore" + " AND channel_id IN (SELECT id FROM channels WHERE NOT from_explore)" + ) + + +def downgrade() -> None: + pass diff --git a/backend/app/catalog/models.py b/backend/app/catalog/models.py index b1bd89e3..5ea760b9 100644 --- a/backend/app/catalog/models.py +++ b/backend/app/catalog/models.py @@ -71,10 +71,12 @@ class Channel(Base, TimestampMixin): Boolean, default=False, server_default="false", index=True ) # True when this channel exists only because a user EXPLORED it (opened its page without - # subscribing) — an ephemeral, per-explorer addition. The cleanup job hard-deletes such a - # channel (and its via_explore videos) once nobody keeps it (no subscription, no live - # ExploredChannel row, no per-user state) after a grace period. Cleared when someone - # subscribes (the "keep" signal promotes it to organic). + # subscribing) — an ephemeral, per-explorer addition. After a grace period the cleanup job + # hard-deletes its via_explore videos nobody keeps (no subscription, no live ExploredChannel + # row, no per-user state), and then the channel row itself once it holds no videos — unless a + # Takeout watch history shows it watched: the "watched but not followed" list names it from + # that row. Cleared when someone subscribes, here or on YouTube (the "keep" signal promotes it + # to organic). from_explore: Mapped[bool] = mapped_column( Boolean, default=False, server_default="false", index=True ) @@ -235,8 +237,9 @@ class ExploredChannel(Base, TimestampMixin): Exploring an un-subscribed channel ingests its recent uploads (flagged via_explore) so the user can browse them, but keeps them out of everyone else's catalog. This row (a) gates the explorer's access to those ephemeral videos in the feed query and (b) drives the grace-period - purge: a from_explore channel with no live ExploredChannel row (and no subscription / no - per-user state on its videos) is hard-deleted by the cleanup job. Subscribing is the "keep" + purge: of a from_explore channel with no live ExploredChannel row (and no subscription) the + cleanup job hard-deletes the videos without per-user state, then the channel once it holds none + and no Takeout watch history shows it watched. Subscribing is the "keep" signal — it promotes the channel to organic and these rows become irrelevant. `created_at` (from TimestampMixin) is the grace clock; a re-visit refreshes it.""" diff --git a/backend/app/catalog/service.py b/backend/app/catalog/service.py index 1004a96e..7567b2a4 100644 --- a/backend/app/catalog/service.py +++ b/backend/app/catalog/service.py @@ -96,13 +96,24 @@ def record_subscription(db: Session, user: User, channel: Channel, yt_subscripti subscribed_at=datetime.now(timezone.utc), ) ) - if channel.from_explore: - channel.from_explore = False - db.execute( - update(Video) - .where(Video.channel_id == channel.id, Video.via_explore.is_(True)) - .values(via_explore=False) - ) + db.flush() + settle_explored(db, channel) + + +def settle_explored(db: Session, channel: Channel) -> None: + """Hold the rule that a channel's videos are an exploration's only while nobody follows it, + decided by who follows it now, not by the flag held in memory: a subscribe in another request + may have cleared that while YouTube answered this one, and the videos that came in meanwhile + would stay flagged. A channel someone follows, however the follow came (here or pulled in from + YouTube), is the catalog's own, and so are its videos.""" + if channel.from_explore and not queries.followed(db, channel.id): + return + channel.from_explore = False + db.execute( + update(Video) + .where(Video.channel_id == channel.id, Video.via_explore.is_(True)) + .values(via_explore=False) + ) db.flush() diff --git a/backend/app/reader/queries.py b/backend/app/reader/queries.py index e2339369..ff090100 100644 --- a/backend/app/reader/queries.py +++ b/backend/app/reader/queries.py @@ -605,7 +605,9 @@ def takeout_lengths_to_ask(db: Session, user: User) -> list[str]: def watched_channels_pending(db: Session, user: User) -> int: """How many channels a later run may still add to the list: the candidates with a watched video - whose length is not known, and the ones that made it but the catalog cannot name yet.""" + whose length is not known, and the ones that made it but have no catalog row to be listed from + yet. One the catalog holds without details is listed already (by the name it came in with), so + it is not one to come.""" cand = _candidate_channels(user) unasked = db.execute( select(TakeoutItem.channel_id) @@ -613,19 +615,30 @@ def watched_channels_pending(db: Session, user: User) -> int: .where(_unfollowed(user), TakeoutItem.looked_up_at.is_(None)) .distinct() ).scalars() - return len(set(unasked) | set(watched_channels_to_describe(db, user))) + counted = _counted_channels(user) + unlisted = db.execute( + select(counted.c.channel_id) + .outerjoin(Channel, Channel.id == counted.c.channel_id) + .where(Channel.id.is_(None)) + ).scalars() + return len(set(unasked) | set(unlisted)) -def watched_channels_to_describe(db: Session, user: User) -> list[str]: - """The channels on the list (by their counted videos) that the catalog lacks or has no details - of yet: what the list cannot name.""" - counted = ( +def _counted_channels(user: User): + """The channels that make the list by their counted videos, whatever the catalog holds of them.""" + return ( select(TakeoutItem.channel_id) .where(_unfollowed(user), TakeoutItem.duration_seconds >= WATCHED_CHANNEL_MIN_SECONDS) .group_by(TakeoutItem.channel_id) .having(func.count() >= WATCHED_CHANNEL_MIN_VIDEOS) .subquery() ) + + +def watched_channels_to_describe(db: Session, user: User) -> list[str]: + """The channels on the list (by their counted videos) that the catalog lacks or has no details + of yet: what the list cannot name, or names with no more than a video brought.""" + counted = _counted_channels(user) return list( db.execute( select(counted.c.channel_id) diff --git a/backend/app/reader/takeout.py b/backend/app/reader/takeout.py index 39382476..301408c2 100644 --- a/backend/app/reader/takeout.py +++ b/backend/app/reader/takeout.py @@ -320,7 +320,10 @@ def stage(db: Session, user: User, parsed: Parsed) -> None: # 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. +# two full Watch later lists (two units a hundred ids). The rest wait for the next day's runs. Both +# shares pace the work, they do not bound its cost: they are held in memory, so a restart starts +# them afresh, which only brings the rest forward — every video is asked about once (a looked-up +# row is marked), and `quota.backfill_reserve_reached` still guards the shared budget. _lookups = RateLimiter(max_events=2 * WATCH_LATER_MAX, window_seconds=24 * 3600) # How many watched videos' lengths one user may ask about in a day for the "watched but not followed" # list (a unit a fifty). A share of its own: a large history asks for thousands, and on the Watch diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index e8fea051..60b68538 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -481,6 +481,8 @@ def watched_channel_videos( ensure_videos(db, yt, missing, explore=channel.from_explore) except YouTubeError as exc: raise youtube_refusal(exc) from exc + catalog_service.settle_explored(db, channel) + db.commit() items = queries.feed_rows_by_ids(db, user, ids) watched_at = dict(counted) for item in items: diff --git a/backend/app/sync/explore.py b/backend/app/sync/explore.py index e6b1b686..f98172cf 100644 --- a/backend/app/sync/explore.py +++ b/backend/app/sync/explore.py @@ -14,6 +14,7 @@ from sqlalchemy import delete, or_, select from sqlalchemy.orm import Session from app import logs, sysconfig +from app.catalog import service as catalog_service from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video from app.reader import queries as reader_queries from app.sync.videos import insert_stubs, stub_from_playlist_item, apply_video_details @@ -91,6 +92,7 @@ def explore_ingest_page( stub["via_explore"] = channel.from_explore stubs.append(stub) inserted = insert_stubs(db, stubs) + catalog_service.settle_explored(db, channel) self_enriched = _enrich_new(db, yt, [s["id"] for s in stubs]) log.info( "explore ingest channel=%s ingested=%s enriched=%s", channel.id, inserted, self_enriched diff --git a/backend/app/sync/subscriptions.py b/backend/app/sync/subscriptions.py index 495af93e..bb37da66 100644 --- a/backend/app/sync/subscriptions.py +++ b/backend/app/sync/subscriptions.py @@ -8,6 +8,7 @@ from sqlalchemy.orm import Session from app import logs from app.models import User +from app.catalog import service as catalog_service from app.catalog.models import Channel, Subscription from app.utils import normalize_description from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail @@ -197,6 +198,10 @@ def import_subscriptions(db: Session, user: User) -> dict: db.add(sub_row) sub_row.yt_subscription_id = sub.get("yt_subscription_id") db.flush() + # A follow made on youtube.com keeps an explored channel as a follow made here does. + explored = select(Channel).where(Channel.id.in_(fetched), Channel.from_explore.is_(True)) + for channel in db.execute(explored).scalars(): + catalog_service.settle_explored(db, channel) # Prune subscriptions removed on YouTube — but only when this fetch is trustworthy # (see `should_prune`). diff --git a/backend/tests/test_catalog_service.py b/backend/tests/test_catalog_service.py index bbc2afc0..a30af2b8 100644 --- a/backend/tests/test_catalog_service.py +++ b/backend/tests/test_catalog_service.py @@ -12,6 +12,7 @@ from datetime import datetime, timedelta, timezone from sqlalchemy import select from app.catalog import queries, search, service +from app.db import SessionLocal from app.catalog.models import ( BlockedChannel, Channel, @@ -23,6 +24,7 @@ from app.catalog.models import ( from app.reader import queries as reader_queries from app.reader import service as reader_service from app.sync import explore as sync_explore +from app.sync import subscriptions as sync_subscriptions def _channel(db, cid="UCc", **kw): @@ -166,6 +168,65 @@ def test_exploring_a_channel_someone_follows_brings_its_uploads_in_as_the_catalo assert (ch.from_explore, db.get(Video, "v1").via_explore) == (False, False) +def test_a_follow_pulled_in_from_youtube_makes_an_explored_channel_the_catalogs_own( + db, make_user, monkeypatch +): + explorer, follower = make_user(), make_user() + ch = _channel(db, details_synced_at=datetime.now(timezone.utc)) + service.explore(db, explorer, ch) + db.commit() + _video(db, "v1", via_explore=True) + + class _Yt: + def __init__(self, session, user): + pass + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def iter_subscriptions(self): + yield {"channel_id": "UCc", "title": "UCc", "yt_subscription_id": "sub-UCc"} + + def get_channels(self, ids): + return [] + + monkeypatch.setattr(sync_subscriptions, "YouTubeClient", _Yt) + sync_subscriptions.import_subscriptions(db, follower) + db.commit() + db.expire_all() + # Followed on youtube.com is followed all the same: left flagged, the channel's later videos + # came in as an exploration's, hidden from the Library and their watched marks waiting for good. + assert (db.get(Channel, "UCc").from_explore, db.get(Video, "v1").via_explore) == (False, False) + + +def test_a_follow_landing_while_an_explored_page_comes_in_leaves_none_of_it_flagged( + db, make_user, monkeypatch +): + explorer, follower = make_user(), make_user() + ch = _channel(db, uploads_playlist_id="UUc") + service.explore(db, explorer, ch) + db.commit() + + class _Yt: + def get_playlist_items(self, playlist_id, page_token): + # Someone subscribes, in a request of their own, while YouTube answers this one. + with SessionLocal() as other: + service.record_subscription(other, follower, other.get(Channel, "UCc"), "sub-UCc") + other.commit() + item = {"contentDetails": {"videoId": "v1", "videoPublishedAt": "2026-05-01T00:00:00Z"}} + return {"items": [item]} + + monkeypatch.setattr(sync_explore, "_enrich_new", lambda session, yt, ids: 0) + sync_explore.explore_ingest_page(db, _Yt(), ch) + db.commit() + db.expire_all() + # The subscribe cleared what was flagged before it; a page that lands after it must not be. + assert (db.get(Channel, "UCc").from_explore, db.get(Video, "v1").via_explore) == (False, False) + + def test_a_block_is_once_and_ends_the_exploration(db, make_user): user = make_user() ch = _channel(db) diff --git a/backend/tests/test_migration_0116.py b/backend/tests/test_migration_0116.py new file mode 100644 index 00000000..813b8ecb --- /dev/null +++ b/backend/tests/test_migration_0116.py @@ -0,0 +1,50 @@ +"""Migration 0116's data step, DB-lane: an explored channel someone already follows becomes the +catalog's own, and no video stays flagged as an exploration's on a channel that is not one. + +The schema half is checked by test_migrations.py; a flag left on is invisible there. +""" + +from alembic import command +from alembic.config import Config +from sqlalchemy import text + +from app.db import engine + + +def test_followed_explored_channels_and_their_videos_are_the_catalogs_own(db, make_user): + follower = make_user().id + cfg = Config("alembic.ini") + command.downgrade(cfg, "0115_takeout_channels") + try: + with engine.begin() as conn: + + def channel(cid, explored): + conn.execute( + text("INSERT INTO channels (id, from_explore) VALUES (:c, :e)"), + {"c": cid, "e": explored}, + ) + conn.execute( + text( + "INSERT INTO videos (id, channel_id, published_at, via_explore)" + " VALUES (:v, :c, now(), true)" + ), + {"v": f"v-{cid}", "c": cid}, + ) + + channel("UCfollowed", True) # explored, then followed on youtube.com + channel("UCexplored", True) # explored, nobody follows it: stays the exploration's + channel("UCorganic", False) # a video flagged after a subscribe had cleared the channel + conn.execute( + text("INSERT INTO subscriptions (user_id, channel_id) VALUES (:u, 'UCfollowed')"), + {"u": follower}, + ) + finally: + command.upgrade(cfg, "head") + + with engine.connect() as conn: + channels = dict(conn.execute(text("SELECT id, from_explore FROM channels")).tuples().all()) + videos = dict( + conn.execute(text("SELECT channel_id, via_explore FROM videos")).tuples().all() + ) + assert channels == {"UCfollowed": False, "UCexplored": True, "UCorganic": False} + assert videos == {"UCfollowed": False, "UCexplored": True, "UCorganic": False} diff --git a/backend/tests/test_watched_channels.py b/backend/tests/test_watched_channels.py index 11bfc3cf..2b76cdba 100644 --- a/backend/tests/test_watched_channels.py +++ b/backend/tests/test_watched_channels.py @@ -10,6 +10,7 @@ from sqlalchemy import select, update from app import quota from app.catalog import service as catalog_service from app.catalog.models import BlockedChannel, Channel, ExploredChannel, Subscription, Video +from app.db import SessionLocal from app.reader import queries, service, takeout from app.reader.models import TakeoutChannelDismissal, TakeoutItem, TakeoutKind from app.routes import feed as feed_routes @@ -318,6 +319,18 @@ def test_naming_the_channels_is_tried_again_and_meanwhile_the_list_says_it_is_no assert queries.watched_channels_pending(db, user) == 0 +def test_a_listed_channel_still_to_be_described_is_not_counted_as_one_to_come(db, make_user): + user = make_user("stub@example.com") + # A stub the catalog made when one of its videos came in: a title, no details yet. + db.add(Channel(id=_ch(1), title="From a video")) + db.commit() + _watched(db, user, 1, [LONG] * 5) + assert queries.watched_channels_to_describe(db, user) == [_ch(1)] + assert _listed(db, user) == [(_ch(1), 5)] + # It is on the list already: "the list may grow" would promise one more that cannot come. + assert queries.watched_channels_pending(db, user) == 0 + + def test_the_lengths_leave_the_watch_later_lookups_their_own_share( db, make_user, youtube, monkeypatch ): @@ -400,6 +413,31 @@ def test_a_followed_channels_watched_videos_come_in_as_the_catalogs_own( assert db.get(Channel, _ch(1)).from_explore is False +def test_a_follow_landing_while_the_videos_come_in_leaves_none_of_them_flagged( + db, make_user, as_user, monkeypatch +): + user, follower = make_user("racer@example.com"), make_user("late@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG] * 5) + fake = _fake_ensure(calls := []) + + def ensure(session, yt, ids, *, explore=False): + # Someone subscribes, in a request of their own, while YouTube answers this one. + with SessionLocal() as other: + ch = other.get(Channel, _ch(1)) + catalog_service.record_subscription(other, follower, ch, "sub-late") + other.commit() + fake(session, yt, ids, explore=explore) + + monkeypatch.setattr(feed_routes, "ensure_videos", ensure) + monkeypatch.setattr(feed_routes.quota, "backfill_reserve_reached", lambda: False) + assert as_user(user).post(f"/api/takeout/channels/{_ch(1)}/videos").status_code == 200 + assert calls[0][1] is True # asked as the exploration's + db.expire_all() + flagged = db.scalars(select(Video.id).where(Video.via_explore.is_(True))).all() + assert (db.get(Channel, _ch(1)).from_explore, flagged) == (False, []) + + def test_the_cleanup_keeps_a_channel_a_watch_history_lists_and_reclaims_its_videos(db, make_user): user = make_user("kept@example.com") _channel(db, 1)