feat: add targeted album list sync background task (#558)

adds sync_album_list task to sync album ATProto list records immediately
after track operations, rather than waiting for the user's next login.

changes:
- add sync_album_list task and schedule_album_list_sync to background_tasks.py
- refactor _register_tasks to use docket.register_collection() per Guidry's advice
- call sync on: track upload, track delete, track album change, record restore
- mock schedule_album_list_sync in affected tests

fixes the issue where albums created during track upload didn't get
ATProto list records until the user logged in again.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude <[e-mail]>

backend/src/backend/_internal/background.py

···
     """register all background task functions with the docket.
 
     tasks must be registered before they can be executed by workers.
-    add new task imports here as they're created.
+    new tasks should be added to background_tasks.background_tasks list.
     """
-    # import task functions here to avoid circular imports
-    from backend._internal.background_tasks import (
-        process_export,
-        scan_copyright,
-        scrobble_to_teal,
-        sync_atproto,
-    )
+    docket.register_collection("backend._internal.background_tasks:background_tasks")
 
-    docket.register(scan_copyright)
-    docket.register(process_export)
-    docket.register(sync_atproto)
-    docket.register(scrobble_to_teal)
-
-    logger.info(
-        "registered background tasks",
-        extra={
-            "tasks": [
-                "scan_copyright",
-                "process_export",
-                "sync_atproto",
-                "scrobble_to_teal",
-            ]
-        },
-    )
+    logger.info("registered background tasks")

backend/src/backend/_internal/background_tasks.py

···
         session_id, track_id, track_title, artist_name, duration, album_name
     )
     logfire.info("scheduled teal scrobble", track_id=track_id)
+
+
+async def sync_album_list(session_id: str, album_id: str) -> None:
+    """sync a single album's ATProto list record.
+
+    creates or updates the album's list record on the user's PDS.
+    called after track uploads or album mutations.
+
+    args:
+        session_id: the user's session ID for authentication
+        album_id: the album's database ID
+    """
+    from sqlalchemy import select
+
+    from backend._internal.atproto.records.fm_plyr import upsert_album_list_record
+    from backend._internal.auth import get_session
+    from backend.models import Album, Track
+    from backend.utilities.database import db_session
+
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"sync_album_list: session {session_id[:8]}... not found")
+        return
+
+    async with db_session() as session:
+        # fetch album
+        album_result = await session.execute(select(Album).where(Album.id == album_id))
+        album = album_result.scalar_one_or_none()
+        if not album:
+            logger.warning(f"sync_album_list: album {album_id} not found")
+            return
+
+        # verify album belongs to this user
+        if album.artist_did != auth_session.did:
+            logger.warning(
+                f"sync_album_list: album {album_id} does not belong to {auth_session.did}"
+            )
+            return
+
+        # fetch tracks with ATProto records
+        tracks_result = await session.execute(
+            select(Track)
+            .where(
+                Track.album_id == album_id,
+                Track.atproto_record_uri.isnot(None),
+                Track.atproto_record_cid.isnot(None),
+            )
+            .order_by(Track.created_at.asc())
+        )
+        tracks = tracks_result.scalars().all()
+
+        if not tracks:
+            logger.debug(
+                f"sync_album_list: album {album_id} has no tracks with ATProto records"
+            )
+            return
+
+        track_refs = [
+            {"uri": t.atproto_record_uri, "cid": t.atproto_record_cid} for t in tracks
+        ]
+
+        try:
+            result = await upsert_album_list_record(
+                auth_session,
+                album_id=album_id,
+                album_title=album.title,
+                track_refs=track_refs,
+                existing_uri=album.atproto_record_uri,
+                existing_created_at=album.created_at,
+            )
+            if result:
+                album.atproto_record_uri = result[0]
+                album.atproto_record_cid = result[1]
+                await session.commit()
+                logger.info(f"synced album list record for {album_id}: {result[0]}")
+        except Exception as e:
+            logger.warning(f"failed to sync album list record for {album_id}: {e}")
+
+
+async def schedule_album_list_sync(session_id: str, album_id: str) -> None:
+    """schedule an album list sync via docket."""
+    docket = get_docket()
+    await docket.add(sync_album_list)(session_id, album_id)
+    logfire.info("scheduled album list sync", album_id=album_id)
+
+
+# collection of all background task functions for docket registration
+background_tasks = [
+    scan_copyright,
+    process_export,
+    sync_atproto,
+    scrobble_to_teal,
+    sync_album_list,
+]

backend/src/backend/api/tracks/mutations.py

···
     update_record,
 )
 from backend._internal.atproto.tid import datetime_to_tid
+from backend._internal.background_tasks import schedule_album_list_sync
 from backend.config import settings
 from backend.models import Artist, Tag, Track, TrackTag, get_db
 from backend.schemas import TrackResponse
···
                     f"failed to delete image {track.image_id}: {e}", exc_info=True
                 )
 
+    # capture album_id before deletion for list sync
+    album_id_to_sync = track.album_id
+
     # delete track record
     await db.delete(track)
     await db.commit()
+
+    # sync album list record if track was in an album
+    if album_id_to_sync:
+        await schedule_album_list_sync(auth_session.session_id, album_id_to_sync)
 
     return {"message": "track deleted successfully"}
 
···
         track.title = title
         title_changed = True
 
+    # track album changes for list sync
+    old_album_id = track.album_id
     await apply_album_update(db, track, album)
+    new_album_id = track.album_id
+    album_changed = old_album_id != new_album_id
 
     if features is not None:
         track.features = await resolve_feature_handles(
···
 
     await db.commit()
     await db.refresh(track)
+
+    # sync album list records if album changed
+    if album_changed:
+        # sync old album (track was removed)
+        if old_album_id:
+            await schedule_album_list_sync(auth_session.session_id, old_album_id)
+        # sync new album (track was added)
+        if new_album_id:
+            await schedule_album_list_sync(auth_session.session_id, new_album_id)
 
     # build track_tags dict for response
     # if tags were updated, use updated_tags; otherwise query for existing
···
     track.atproto_record_cid = new_cid
     await db.commit()
     await db.refresh(track)
+
+    # sync album list if track is in an album
+    if track.album_id:
+        await schedule_album_list_sync(auth_session.session_id, track.album_id)
 
     logger.info(f"restored ATProto record for track {track_id}: {new_uri}")
 

backend/src/backend/api/tracks/uploads.py

···
 from backend._internal.atproto import create_track_record
 from backend._internal.atproto.handles import resolve_featured_artists
 from backend._internal.audio import AudioFormat
-from backend._internal.background_tasks import schedule_copyright_scan
+from backend._internal.background_tasks import (
+    schedule_album_list_sync,
+    schedule_copyright_scan,
+)
 from backend._internal.image import ImageFormat
 from backend._internal.jobs import job_service
 from backend.config import settings
···
 
                     if r2_url:
                         await schedule_copyright_scan(track.id, r2_url)
+
+                    # sync album list record if track is in an album
+                    if album_record:
+                        await schedule_album_list_sync(
+                            ctx.auth_session.session_id, album_record.id
+                        )
 
                     await job_service.update_progress(
                         ctx.upload_id,

backend/tests/api/test_track_deletion.py

···
             "backend.api.tracks.mutations.storage.delete",
             side_effect=mock_delete,
         ),
+        patch(
+            "backend.api.tracks.mutations.schedule_album_list_sync",
+            new_callable=AsyncMock,
+        ),
     ):
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
···
         patch(
             "backend.api.tracks.mutations.storage.delete",
             side_effect=mock_delete,
+        ),
+        patch(
+            "backend.api.tracks.mutations.schedule_album_list_sync",
+            new_callable=AsyncMock,
         ),
     ):
         async with AsyncClient(