feat: move like and comment PDS writes to background tasks (#561)

* feat: move like and comment PDS writes to background tasks

Likes:
- Update like/unlike endpoints to use optimistic DB writes
- Schedule PDS record operations via docket background tasks
- Make atproto_like_uri nullable to support async PDS writes
- Update tests to verify background task scheduling

Comments:
- Update create/update/delete comment endpoints to use optimistic DB writes
- Schedule PDS record operations via docket background tasks
- Make atproto_comment_uri nullable to support async PDS writes
- Update tests to verify background task scheduling

This reduces API response times by moving slow PDS writes to the background
while keeping the UI responsive with immediate local database updates.

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

Co-Authored-By: Claude <[e-mail]>

* refactor: move deferred imports to module level

Move imports that were unnecessarily deferred inside PDS task functions
to the top of the module for clarity.

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

Co-Authored-By: Claude <[e-mail]>

---------

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

.claude/commands/investigate-report.md

···
+---
+description: Investigate the user report and fix it if you find it's a bug.
+argumment-hint: [the user report]
+---
+
+First read the user report: $ARGUMENTS
+
+Look around the codebase via `tree`, `rg` and perhaps related git history to find the root cause of the bug.
+
+If you find the root cause, write a regression test, watch the test fail, fix the bug, and watch the test pass, then open a PR with the fix.

backend/alembic/versions/2025_12_10_125948_4febe85be5a4_allow_null_atproto_like_uri_for_.py

···
+"""allow null atproto_like_uri for optimistic writes
+
+Revision ID: 4febe85be5a4
+Revises: a6069b752a90
+Create Date: 2025-12-10 12:59:48.646844
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "4febe85be5a4"
+down_revision: str | Sequence[str] | None = "a6069b752a90"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Make atproto_like_uri nullable for background PDS writes."""
+    op.alter_column(
+        "track_likes",
+        "atproto_like_uri",
+        existing_type=sa.VARCHAR(),
+        nullable=True,
+    )
+
+
+def downgrade() -> None:
+    """Revert atproto_like_uri to non-nullable."""
+    # first, delete any rows with NULL atproto_like_uri
+    op.execute("DELETE FROM track_likes WHERE atproto_like_uri IS NULL")
+    op.alter_column(
+        "track_likes",
+        "atproto_like_uri",
+        existing_type=sa.VARCHAR(),
+        nullable=False,
+    )

backend/alembic/versions/2025_12_10_131338_37cc1d6980c3_allow_null_atproto_comment_uri_for_.py

···
+"""allow null atproto_comment_uri for optimistic writes
+
+Revision ID: 37cc1d6980c3
+Revises: 4febe85be5a4
+Create Date: 2025-12-10 13:13:38.218675
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "37cc1d6980c3"
+down_revision: str | Sequence[str] | None = "4febe85be5a4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Make atproto_comment_uri nullable for background PDS writes."""
+    op.alter_column(
+        "track_comments",
+        "atproto_comment_uri",
+        existing_type=sa.VARCHAR(),
+        nullable=True,
+    )
+
+
+def downgrade() -> None:
+    """Revert atproto_comment_uri to non-nullable."""
+    # first, delete any rows with NULL atproto_comment_uri
+    op.execute("DELETE FROM track_comments WHERE atproto_comment_uri IS NULL")
+    op.alter_column(
+        "track_comments",
+        "atproto_comment_uri",
+        existing_type=sa.VARCHAR(),
+        nullable=False,
+    )

backend/src/backend/_internal/background_tasks.py

···
 import os
 import tempfile
 import zipfile
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 
 import aioboto3
 import aiofiles
 import logfire
+from sqlalchemy import select
 
+from backend._internal.atproto.records import (
+    create_comment_record,
+    create_like_record,
+    delete_record_by_uri,
+    update_comment_record,
+)
+from backend._internal.auth import get_session
 from backend._internal.background import get_docket
+from backend.models import TrackComment, TrackLike
+from backend.utilities.database import db_session
 
 logger = logging.getLogger(__name__)
 
···
     logfire.info("scheduled album list sync", album_id=album_id)
 
 
+# ---------------------------------------------------------------------------
+# PDS record write tasks
+#
+# these tasks handle writing records to the user's PDS (Personal Data Server)
+# in the background, then updating the local database with the result.
+# this keeps API responses fast while ensuring PDS and DB stay in sync.
+# ---------------------------------------------------------------------------
+
+
+async def pds_create_like(
+    session_id: str,
+    like_id: int,
+    subject_uri: str,
+    subject_cid: str,
+) -> None:
+    """create a like record on the user's PDS and update the database.
+
+    args:
+        session_id: the user's session ID for authentication
+        like_id: database ID of the TrackLike record to update
+        subject_uri: AT URI of the track being liked
+        subject_cid: CID of the track being liked
+    """
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"pds_create_like: session {session_id[:8]}... not found")
+        return
+
+    try:
+        like_uri = await create_like_record(
+            auth_session=auth_session,
+            subject_uri=subject_uri,
+            subject_cid=subject_cid,
+        )
+
+        # update database with the ATProto URI
+        async with db_session() as session:
+            result = await session.execute(
+                select(TrackLike).where(TrackLike.id == like_id)
+            )
+            like = result.scalar_one_or_none()
+            if like:
+                like.atproto_like_uri = like_uri
+                await session.commit()
+                logger.info(f"pds_create_like: created like record {like_uri}")
+            else:
+                # like was deleted before we could update it - clean up orphan
+                logger.warning(f"pds_create_like: like {like_id} no longer exists")
+                await delete_record_by_uri(auth_session, like_uri)
+
+    except Exception as e:
+        logger.error(f"pds_create_like failed for like {like_id}: {e}", exc_info=True)
+        # note: we don't delete the DB record on failure - user still sees "liked"
+        # and we can retry or fix later. this is better than inconsistent state.
+
+
+async def schedule_pds_create_like(
+    session_id: str,
+    like_id: int,
+    subject_uri: str,
+    subject_cid: str,
+) -> None:
+    """schedule a like record creation via docket."""
+    docket = get_docket()
+    await docket.add(pds_create_like)(session_id, like_id, subject_uri, subject_cid)
+    logfire.info("scheduled pds like creation", like_id=like_id)
+
+
+async def pds_delete_like(
+    session_id: str,
+    like_uri: str,
+) -> None:
+    """delete a like record from the user's PDS.
+
+    args:
+        session_id: the user's session ID for authentication
+        like_uri: AT URI of the like record to delete
+    """
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"pds_delete_like: session {session_id[:8]}... not found")
+        return
+
+    try:
+        await delete_record_by_uri(auth_session, like_uri)
+        logger.info(f"pds_delete_like: deleted like record {like_uri}")
+    except Exception as e:
+        logger.error(f"pds_delete_like failed for {like_uri}: {e}", exc_info=True)
+        # deletion failed - the PDS record may still exist, but DB is already clean
+        # this is acceptable: orphaned PDS records are harmless
+
+
+async def schedule_pds_delete_like(session_id: str, like_uri: str) -> None:
+    """schedule a like record deletion via docket."""
+    docket = get_docket()
+    await docket.add(pds_delete_like)(session_id, like_uri)
+    logfire.info("scheduled pds like deletion", like_uri=like_uri)
+
+
+async def pds_create_comment(
+    session_id: str,
+    comment_id: int,
+    subject_uri: str,
+    subject_cid: str,
+    text: str,
+    timestamp_ms: int,
+) -> None:
+    """create a comment record on the user's PDS and update the database.
+
+    args:
+        session_id: the user's session ID for authentication
+        comment_id: database ID of the TrackComment record to update
+        subject_uri: AT URI of the track being commented on
+        subject_cid: CID of the track being commented on
+        text: comment text
+        timestamp_ms: playback position when comment was made
+    """
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"pds_create_comment: session {session_id[:8]}... not found")
+        return
+
+    try:
+        comment_uri = await create_comment_record(
+            auth_session=auth_session,
+            subject_uri=subject_uri,
+            subject_cid=subject_cid,
+            text=text,
+            timestamp_ms=timestamp_ms,
+        )
+
+        # update database with the ATProto URI
+        async with db_session() as session:
+            result = await session.execute(
+                select(TrackComment).where(TrackComment.id == comment_id)
+            )
+            comment = result.scalar_one_or_none()
+            if comment:
+                comment.atproto_comment_uri = comment_uri
+                await session.commit()
+                logger.info(f"pds_create_comment: created comment record {comment_uri}")
+            else:
+                # comment was deleted before we could update it - clean up orphan
+                logger.warning(
+                    f"pds_create_comment: comment {comment_id} no longer exists"
+                )
+                await delete_record_by_uri(auth_session, comment_uri)
+
+    except Exception as e:
+        logger.error(
+            f"pds_create_comment failed for comment {comment_id}: {e}", exc_info=True
+        )
+
+
+async def schedule_pds_create_comment(
+    session_id: str,
+    comment_id: int,
+    subject_uri: str,
+    subject_cid: str,
+    text: str,
+    timestamp_ms: int,
+) -> None:
+    """schedule a comment record creation via docket."""
+    docket = get_docket()
+    await docket.add(pds_create_comment)(
+        session_id, comment_id, subject_uri, subject_cid, text, timestamp_ms
+    )
+    logfire.info("scheduled pds comment creation", comment_id=comment_id)
+
+
+async def pds_delete_comment(
+    session_id: str,
+    comment_uri: str,
+) -> None:
+    """delete a comment record from the user's PDS.
+
+    args:
+        session_id: the user's session ID for authentication
+        comment_uri: AT URI of the comment record to delete
+    """
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"pds_delete_comment: session {session_id[:8]}... not found")
+        return
+
+    try:
+        await delete_record_by_uri(auth_session, comment_uri)
+        logger.info(f"pds_delete_comment: deleted comment record {comment_uri}")
+    except Exception as e:
+        logger.error(f"pds_delete_comment failed for {comment_uri}: {e}", exc_info=True)
+
+
+async def schedule_pds_delete_comment(session_id: str, comment_uri: str) -> None:
+    """schedule a comment record deletion via docket."""
+    docket = get_docket()
+    await docket.add(pds_delete_comment)(session_id, comment_uri)
+    logfire.info("scheduled pds comment deletion", comment_uri=comment_uri)
+
+
+async def pds_update_comment(
+    session_id: str,
+    comment_id: int,
+    comment_uri: str,
+    subject_uri: str,
+    subject_cid: str,
+    text: str,
+    timestamp_ms: int,
+    created_at: datetime,
+) -> None:
+    """update a comment record on the user's PDS.
+
+    args:
+        session_id: the user's session ID for authentication
+        comment_id: database ID of the TrackComment record
+        comment_uri: AT URI of the comment record to update
+        subject_uri: AT URI of the track being commented on
+        subject_cid: CID of the track being commented on
+        text: new comment text
+        timestamp_ms: playback position when comment was made
+        created_at: original creation timestamp
+    """
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"pds_update_comment: session {session_id[:8]}... not found")
+        return
+
+    try:
+        await update_comment_record(
+            auth_session=auth_session,
+            comment_uri=comment_uri,
+            subject_uri=subject_uri,
+            subject_cid=subject_cid,
+            text=text,
+            timestamp_ms=timestamp_ms,
+            created_at=created_at,
+            updated_at=datetime.now(UTC),
+        )
+        logger.info(f"pds_update_comment: updated comment record {comment_uri}")
+    except Exception as e:
+        logger.error(
+            f"pds_update_comment failed for comment {comment_id}: {e}", exc_info=True
+        )
+
+
+async def schedule_pds_update_comment(
+    session_id: str,
+    comment_id: int,
+    comment_uri: str,
+    subject_uri: str,
+    subject_cid: str,
+    text: str,
+    timestamp_ms: int,
+    created_at: datetime,
+) -> None:
+    """schedule a comment record update via docket."""
+    docket = get_docket()
+    await docket.add(pds_update_comment)(
+        session_id,
+        comment_id,
+        comment_uri,
+        subject_uri,
+        subject_cid,
+        text,
+        timestamp_ms,
+        created_at,
+    )
+    logfire.info("scheduled pds comment update", comment_id=comment_id)
+
+
 # collection of all background task functions for docket registration
 background_tasks = [
     scan_copyright,
···
     sync_atproto,
     scrobble_to_teal,
     sync_album_list,
+    pds_create_like,
+    pds_delete_like,
+    pds_create_comment,
+    pds_delete_comment,
+    pds_update_comment,
 ]

backend/src/backend/api/tracks/comments.py

···
 from datetime import UTC, datetime
 from typing import Annotated
 
-from fastapi import Depends, HTTPException
+from fastapi import Cookie, Depends, HTTPException, Request
 from pydantic import BaseModel, Field
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend._internal import Session as AuthSession
 from backend._internal import require_auth
-from backend._internal.atproto import (
-    create_comment_record,
-    delete_record_by_uri,
-    update_comment_record,
+from backend._internal.background_tasks import (
+    schedule_pds_create_comment,
+    schedule_pds_delete_comment,
+    schedule_pds_update_comment,
 )
 from backend.models import Artist, Track, TrackComment, UserPreferences, get_db
 
···
 async def create_comment(
     track_id: int,
     body: CommentCreate,
+    request: Request,
     db: Annotated[AsyncSession, Depends(get_db)],
     auth_session: AuthSession = Depends(require_auth),
+    session_id_cookie: Annotated[str | None, Cookie(alias="session_id")] = None,
 ) -> CommentResponse:
     """create a timed comment on a track.
 
     requires auth. track owner must have allow_comments enabled.
+    the comment is visible immediately; the ATProto record is created in background.
     """
     # get track
     track_result = await db.execute(select(Track).where(Track.id == track_id))
···
             detail=f"track has reached maximum of {MAX_COMMENTS_PER_TRACK} comments",
         )
 
-    # create ATProto record
-    comment_uri = None
-    try:
-        comment_uri = await create_comment_record(
-            auth_session=auth_session,
-            subject_uri=track.atproto_record_uri,
-            subject_cid=track.atproto_record_cid,
-            text=body.text,
-            timestamp_ms=body.timestamp_ms,
-        )
-
-        # create database record
-        comment = TrackComment(
-            track_id=track_id,
-            user_did=auth_session.did,
-            text=body.text,
-            timestamp_ms=body.timestamp_ms,
-            atproto_comment_uri=comment_uri,
-        )
-        db.add(comment)
-        await db.commit()
-        await db.refresh(comment)
+    # create database record immediately (optimistic)
+    comment = TrackComment(
+        track_id=track_id,
+        user_did=auth_session.did,
+        text=body.text,
+        timestamp_ms=body.timestamp_ms,
+        atproto_comment_uri=None,  # will be set by background task
+    )
+    db.add(comment)
+    await db.commit()
+    await db.refresh(comment)
 
-    except Exception as e:
-        await db.rollback()
-        logger.error(
-            f"failed to create comment on track {track_id} by {auth_session.did}: {e}",
-            exc_info=True,
-        )
-        # cleanup ATProto record if created
-        if comment_uri:
-            try:
-                await delete_record_by_uri(auth_session, comment_uri)
-                logger.info(
-                    f"cleaned up orphaned ATProto comment record: {comment_uri}"
-                )
-            except Exception as cleanup_exc:
-                logger.error(
-                    f"failed to cleanup orphaned comment record: {cleanup_exc}"
-                )
-        raise HTTPException(status_code=500, detail="failed to create comment") from e
+    # schedule PDS record creation in background
+    session_id = session_id_cookie or request.headers.get("authorization", "").replace(
+        "Bearer ", ""
+    )
+    await schedule_pds_create_comment(
+        session_id=session_id,
+        comment_id=comment.id,
+        subject_uri=track.atproto_record_uri,
+        subject_cid=track.atproto_record_cid,
+        text=body.text,
+        timestamp_ms=body.timestamp_ms,
+    )
 
     # get user info for response
     artist_result = await db.execute(
···
 @router.delete("/comments/{comment_id}")
 async def delete_comment(
     comment_id: int,
+    request: Request,
     db: Annotated[AsyncSession, Depends(get_db)],
     auth_session: AuthSession = Depends(require_auth),
+    session_id_cookie: Annotated[str | None, Cookie(alias="session_id")] = None,
 ) -> dict:
-    """delete a comment. only the author can delete their own comments."""
+    """delete a comment. only the author can delete their own comments.
+
+    the comment is removed immediately; the ATProto record is deleted in background.
+    """
     comment_result = await db.execute(
         select(TrackComment).where(TrackComment.id == comment_id)
     )
···
     if comment.user_did != auth_session.did:
         raise HTTPException(status_code=403, detail="can only delete your own comments")
 
-    # delete ATProto record
-    try:
-        await delete_record_by_uri(auth_session, comment.atproto_comment_uri)
-    except Exception as e:
-        logger.error(f"failed to delete ATProto comment record: {e}")
-        # continue with DB deletion anyway
+    # capture the ATProto URI before deleting the DB record
+    comment_uri = comment.atproto_comment_uri
 
+    # delete database record immediately (optimistic)
     await db.delete(comment)
     await db.commit()
 
+    # schedule PDS record deletion in background (if URI exists)
+    if comment_uri:
+        session_id = session_id_cookie or request.headers.get(
+            "authorization", ""
+        ).replace("Bearer ", "")
+        await schedule_pds_delete_comment(
+            session_id=session_id,
+            comment_uri=comment_uri,
+        )
+
     return {"deleted": True}
 
 
···
 async def update_comment(
     comment_id: int,
     body: CommentUpdate,
+    request: Request,
     db: Annotated[AsyncSession, Depends(get_db)],
     auth_session: AuthSession = Depends(require_auth),
+    session_id_cookie: Annotated[str | None, Cookie(alias="session_id")] = None,
 ) -> CommentResponse:
-    """update a comment's text. only the author can edit their own comments."""
+    """update a comment's text. only the author can edit their own comments.
+
+    the comment is updated immediately; the ATProto record is updated in background.
+    """
     comment_result = await db.execute(
         select(TrackComment).where(TrackComment.id == comment_id)
     )
···
             detail="track missing ATProto record - cannot update comment",
         )
 
-    # update ATProto record first
+    # update database record immediately (optimistic)
     updated_at = datetime.now(UTC)
-    try:
-        await update_comment_record(
-            auth_session=auth_session,
+    comment.text = body.text
+    comment.updated_at = updated_at
+
+    await db.commit()
+    await db.refresh(comment)
+
+    # schedule PDS record update in background (if URI exists)
+    if comment.atproto_comment_uri:
+        session_id = session_id_cookie or request.headers.get(
+            "authorization", ""
+        ).replace("Bearer ", "")
+        await schedule_pds_update_comment(
+            session_id=session_id,
+            comment_id=comment.id,
             comment_uri=comment.atproto_comment_uri,
             subject_uri=track.atproto_record_uri,
             subject_cid=track.atproto_record_cid,
             text=body.text,
             timestamp_ms=comment.timestamp_ms,
             created_at=comment.created_at,
-            updated_at=updated_at,
         )
-    except Exception as e:
-        logger.error(
-            f"failed to update ATProto comment record {comment.atproto_comment_uri}: {e}",
-            exc_info=True,
-        )
-        raise HTTPException(
-            status_code=500, detail="failed to update comment on ATProto"
-        ) from e
-
-    # update database record
-    comment.text = body.text
-    comment.updated_at = updated_at
-
-    await db.commit()
-    await db.refresh(comment)
 
     # get user info for response
     artist_result = await db.execute(

backend/src/backend/api/tracks/likes.py

···
 import logging
 from typing import Annotated
 
-from fastapi import Depends, HTTPException
+from fastapi import Cookie, Depends, HTTPException, Request
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from backend._internal import Session as AuthSession
 from backend._internal import require_auth
-from backend._internal.atproto import create_like_record, delete_record_by_uri
+from backend._internal.background_tasks import (
+    schedule_pds_create_like,
+    schedule_pds_delete_like,
+)
 from backend.models import Artist, Track, TrackLike, get_db
 from backend.schemas import TrackResponse
 from backend.utilities.aggregations import get_comment_counts, get_like_counts
···
 @router.post("/{track_id}/like")
 async def like_track(
     track_id: int,
+    request: Request,
     db: Annotated[AsyncSession, Depends(get_db)],
     auth_session: AuthSession = Depends(require_auth),
+    session_id_cookie: Annotated[str | None, Cookie(alias="session_id")] = None,
 ) -> dict:
-    """Like a track - creates ATProto record and stores in index."""
+    """Like a track - stores in database immediately, creates ATProto record in background.
+
+    The like is visible immediately in the UI. The ATProto record is created
+    asynchronously via a background task, keeping the API response fast.
+    """
     result = await db.execute(select(Track).where(Track.id == track_id))
     track = result.scalar_one_or_none()
 
···
     if existing_like.scalar_one_or_none():
         return {"liked": True}
 
-    like_uri = None
-    try:
-        like_uri = await create_like_record(
-            auth_session=auth_session,
-            subject_uri=track.atproto_record_uri,
-            subject_cid=track.atproto_record_cid,
-        )
+    # create database record immediately (optimistic)
+    like = TrackLike(
+        track_id=track_id,
+        user_did=auth_session.did,
+        atproto_like_uri=None,  # will be set by background task
+    )
+    db.add(like)
+    await db.commit()
+    await db.refresh(like)
 
-        like = TrackLike(
-            track_id=track_id,
-            user_did=auth_session.did,
-            atproto_like_uri=like_uri,
-        )
-        db.add(like)
-        await db.commit()
-    except Exception as e:
-        logger.error(
-            f"failed to like track {track_id} for user {auth_session.did}: {e}",
-            exc_info=True,
-        )
-        if like_uri:
-            try:
-                await delete_record_by_uri(
-                    auth_session=auth_session,
-                    record_uri=like_uri,
-                )
-                logger.info(f"cleaned up orphaned ATProto like record: {like_uri}")
-            except Exception as cleanup_exc:
-                logger.error(
-                    f"failed to clean up orphaned ATProto like record {like_uri}: {cleanup_exc}"
-                )
-        raise HTTPException(
-            status_code=500, detail="failed to like track - please try again"
-        ) from e
+    # schedule PDS record creation in background
+    session_id = session_id_cookie or request.headers.get("authorization", "").replace(
+        "Bearer ", ""
+    )
+    await schedule_pds_create_like(
+        session_id=session_id,
+        like_id=like.id,
+        subject_uri=track.atproto_record_uri,
+        subject_cid=track.atproto_record_cid,
+    )
 
-    return {"liked": True, "atproto_uri": like_uri}
+    return {"liked": True}
 
 
 @router.delete("/{track_id}/like")
 async def unlike_track(
     track_id: int,
+    request: Request,
     db: Annotated[AsyncSession, Depends(get_db)],
     auth_session: AuthSession = Depends(require_auth),
+    session_id_cookie: Annotated[str | None, Cookie(alias="session_id")] = None,
 ) -> dict:
-    """Unlike a track - deletes ATProto record and removes from index."""
+    """Unlike a track - removes from database immediately, deletes ATProto record in background.
+
+    The unlike is reflected immediately in the UI. The ATProto record deletion
+    happens asynchronously via a background task.
+    """
     result = await db.execute(
         select(TrackLike).where(
             TrackLike.track_id == track_id, TrackLike.user_did == auth_session.did
···
     if not like:
         return {"liked": False}
 
-    track_result = await db.execute(select(Track).where(Track.id == track_id))
-    track = track_result.scalar_one_or_none()
-    if not track:
-        raise HTTPException(status_code=404, detail="track not found")
+    # capture the ATProto URI before deleting the DB record
+    like_uri = like.atproto_like_uri
 
-    if not track.atproto_record_uri or not track.atproto_record_cid:
-        raise HTTPException(
-            status_code=422,
-            detail={
-                "error": "missing_atproto_record",
-                "message": "this track cannot be unliked because its ATProto record is missing",
-            },
-        )
+    # delete database record immediately (optimistic)
+    await db.delete(like)
+    await db.commit()
 
-    await delete_record_by_uri(
-        auth_session=auth_session,
-        record_uri=like.atproto_like_uri,
-    )
-
-    await db.delete(like)
-    try:
-        await db.commit()
-    except Exception as e:
-        logger.error(
-            f"failed to commit unlike to database after deleting ATProto record: {e}"
+    # schedule PDS record deletion in background (if URI exists)
+    if like_uri:
+        session_id = session_id_cookie or request.headers.get(
+            "authorization", ""
+        ).replace("Bearer ", "")
+        await schedule_pds_delete_like(
+            session_id=session_id,
+            like_uri=like_uri,
         )
-        try:
-            recreated_uri = await create_like_record(
-                auth_session=auth_session,
-                subject_uri=track.atproto_record_uri,
-                subject_cid=track.atproto_record_cid,
-            )
-            logger.info(
-                f"rolled back ATProto deletion by recreating like: {recreated_uri}"
-            )
-        except Exception as rollback_exc:
-            logger.critical(
-                f"failed to rollback ATProto deletion for track {track_id}, "
-                f"user {auth_session.did}: {rollback_exc}. "
-                "database and ATProto are now inconsistent"
-            )
-        raise HTTPException(
-            status_code=500, detail="failed to unlike track - please try again"
-        ) from e
 
     return {"liked": False}
 

backend/src/backend/models/track_comment.py

···
     timestamp_ms: Mapped[int] = mapped_column(Integer, nullable=False)
 
     # ATProto comment record URI (source of truth)
-    atproto_comment_uri: Mapped[str] = mapped_column(
-        String, nullable=False, unique=True, index=True
+    # nullable to support optimistic DB writes before PDS record creation
+    atproto_comment_uri: Mapped[str | None] = mapped_column(
+        String, nullable=True, unique=True, index=True
     )
 
     # when it was created (indexed from ATProto record)

backend/src/backend/models/track_like.py

···
     user_did: Mapped[str] = mapped_column(String, nullable=False, index=True)
 
     # ATProto like record URI (source of truth)
-    atproto_like_uri: Mapped[str] = mapped_column(
-        String, nullable=False, unique=True, index=True
+    # nullable to support optimistic DB writes before PDS record creation
+    atproto_like_uri: Mapped[str | None] = mapped_column(
+        String, nullable=True, unique=True, index=True
     )
 
     # when it was liked (indexed from ATProto record)

backend/tests/api/test_track_comments.py

···
     artist_with_comments_enabled: UserPreferences,
     commenter_artist: Artist,
 ):
-    """test successful comment creation."""
-    with patch("backend.api.tracks.comments.create_comment_record") as mock_create:
-        mock_create.return_value = "at://did:test:commenter123/fm.plyr.comment/xyz"
-
+    """test successful comment creation schedules background task."""
+    with patch(
+        "backend.api.tracks.comments.schedule_pds_create_comment"
+    ) as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert data["timestamp_ms"] == 30000
     assert data["user_did"] == "did:test:commenter123"
 
-    # verify ATProto record was created
-    mock_create.assert_called_once()
+    # verify background task was scheduled
+    mock_schedule.assert_called_once()
+    call_kwargs = mock_schedule.call_args.kwargs
+    assert call_kwargs["subject_uri"] == test_track.atproto_record_uri
+    assert call_kwargs["subject_cid"] == test_track.atproto_record_cid
+    assert call_kwargs["text"] == "awesome drop at this moment!"
+    assert call_kwargs["timestamp_ms"] == 30000
 
-    # verify DB entry
+    # verify DB entry exists (created immediately, before PDS)
     result = await db_session.execute(
         select(TrackComment).where(TrackComment.track_id == test_track.id)
     )
     comment = result.scalar_one()
     assert comment.text == "awesome drop at this moment!"
     assert comment.timestamp_ms == 30000
+    # atproto_comment_uri is None initially - will be set by background task
+    assert comment.atproto_comment_uri is None
+
+
+async def test_create_comment_db_entry_has_correct_comment_id(
+    test_app: FastAPI,
+    db_session: AsyncSession,
+    test_track: Track,
+    artist_with_comments_enabled: UserPreferences,
+    commenter_artist: Artist,
+):
+    """test that the comment_id passed to background task matches the DB record."""
+    with patch(
+        "backend.api.tracks.comments.schedule_pds_create_comment"
+    ) as mock_schedule:
+        async with AsyncClient(
+            transport=ASGITransport(app=test_app), base_url="http://test"
+        ) as client:
+            await client.post(
+                f"/tracks/{test_track.id}/comments",
+                json={"text": "test comment", "timestamp_ms": 5000},
+            )
+
+    # get the comment_id from the scheduled call
+    call_kwargs = mock_schedule.call_args.kwargs
+    scheduled_comment_id = call_kwargs["comment_id"]
+
+    # verify it matches the DB record
+    result = await db_session.execute(
+        select(TrackComment).where(TrackComment.track_id == test_track.id)
+    )
+    comment = result.scalar_one()
+    assert comment.id == scheduled_comment_id
 
 
 async def test_create_comment_respects_limit(
···
     artist_with_comments_enabled: UserPreferences,
     commenter_artist: Artist,
 ):
-    """test that comment owner can edit their comment and ATProto record is updated."""
-    from unittest.mock import AsyncMock, patch
-
+    """test that comment owner can edit their comment and background task is scheduled."""
     comment = TrackComment(
         track_id=test_track.id,
         user_did="did:test:commenter123",
···
     await db_session.refresh(comment)
 
     with patch(
-        "backend.api.tracks.comments.update_comment_record", new_callable=AsyncMock
-    ) as mock_update:
-        mock_update.return_value = "bafynewcid"
-
+        "backend.api.tracks.comments.schedule_pds_update_comment"
+    ) as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert data["text"] == "edited text"
     assert data["updated_at"] is not None
 
-    # verify ATProto record was updated
-    mock_update.assert_called_once()
-    call_kwargs = mock_update.call_args.kwargs
+    # verify background task was scheduled
+    mock_schedule.assert_called_once()
+    call_kwargs = mock_schedule.call_args.kwargs
     assert call_kwargs["comment_uri"] == comment.atproto_comment_uri
     assert call_kwargs["subject_uri"] == test_track.atproto_record_uri
     assert call_kwargs["subject_cid"] == test_track.atproto_record_cid
···
     assert call_kwargs["timestamp_ms"] == 5000
 
 
-async def test_edit_comment_syncs_to_atproto(
+async def test_edit_comment_without_atproto_uri(
     test_app: FastAPI,
     db_session: AsyncSession,
     test_track: Track,
     artist_with_comments_enabled: UserPreferences,
     commenter_artist: Artist,
 ):
-    """regression test: editing a comment must update the ATProto record."""
-    from unittest.mock import AsyncMock, patch
-
+    """test that editing a comment without ATProto URI doesn't schedule update."""
+    # comment without ATProto URI (e.g., background task hasn't run yet)
     comment = TrackComment(
         track_id=test_track.id,
         user_did="did:test:commenter123",
         text="original text",
-        timestamp_ms=12345,
-        atproto_comment_uri="at://did:test:commenter123/fm.plyr.comment/sync1",
+        timestamp_ms=5000,
+        atproto_comment_uri=None,
     )
     db_session.add(comment)
     await db_session.commit()
     await db_session.refresh(comment)
-    original_created_at = comment.created_at
 
     with patch(
-        "backend.api.tracks.comments.update_comment_record", new_callable=AsyncMock
-    ) as mock_update:
-        mock_update.return_value = "bafynewcid123"
-
+        "backend.api.tracks.comments.schedule_pds_update_comment"
+    ) as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
             response = await client.patch(
                 f"/tracks/comments/{comment.id}",
-                json={"text": "updated via edit"},
+                json={"text": "edited text"},
             )
 
     assert response.status_code == 200
+    data = response.json()
+    assert data["text"] == "edited text"
 
-    # verify ATProto sync happened with correct data
-    mock_update.assert_called_once()
-    call_kwargs = mock_update.call_args.kwargs
-
-    # subject must reference the track's ATProto record
-    assert call_kwargs["subject_uri"] == test_track.atproto_record_uri
-    assert call_kwargs["subject_cid"] == test_track.atproto_record_cid
-
-    # original timestamp_ms and created_at preserved
-    assert call_kwargs["timestamp_ms"] == 12345
-    assert call_kwargs["created_at"] == original_created_at
-
-    # new text and updated_at passed
-    assert call_kwargs["text"] == "updated via edit"
-    assert "updated_at" in call_kwargs  # updated_at should be set
+    # no PDS update should be scheduled since there's no ATProto record
+    mock_schedule.assert_not_called()
 
 
 async def test_edit_comment_forbidden_for_other_user(
···
     artist_with_comments_enabled: UserPreferences,
     commenter_artist: Artist,
 ):
-    """test that comment owner can delete their comment."""
-    from unittest.mock import AsyncMock, patch
-
+    """test that comment owner can delete their comment and background task is scheduled."""
     comment = TrackComment(
         track_id=test_track.id,
         user_did="did:test:commenter123",
···
     comment_id = comment.id
 
     with patch(
-        "backend.api.tracks.comments.delete_record_by_uri", new_callable=AsyncMock
-    ):
+        "backend.api.tracks.comments.schedule_pds_delete_comment"
+    ) as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert response.status_code == 200
     assert response.json()["deleted"] is True
 
+    # verify background task was scheduled with correct URI
+    mock_schedule.assert_called_once()
+    call_kwargs = mock_schedule.call_args.kwargs
+    assert (
+        call_kwargs["comment_uri"] == "at://did:test:commenter123/fm.plyr.comment/del1"
+    )
+
+    # verify DB entry is gone (deleted immediately, before PDS)
+    result = await db_session.execute(
+        select(TrackComment).where(TrackComment.id == comment_id)
+    )
+    assert result.scalar_one_or_none() is None
+
+
+async def test_delete_comment_without_atproto_uri(
+    test_app: FastAPI,
+    db_session: AsyncSession,
+    test_track: Track,
+    artist_with_comments_enabled: UserPreferences,
+    commenter_artist: Artist,
+):
+    """test that deleting a comment without ATProto URI doesn't schedule deletion."""
+    # comment without ATProto URI (e.g., background task hasn't run yet)
+    comment = TrackComment(
+        track_id=test_track.id,
+        user_did="did:test:commenter123",
+        text="to be deleted",
+        timestamp_ms=5000,
+        atproto_comment_uri=None,
+    )
+    db_session.add(comment)
+    await db_session.commit()
+    await db_session.refresh(comment)
+    comment_id = comment.id
+
+    with patch(
+        "backend.api.tracks.comments.schedule_pds_delete_comment"
+    ) as mock_schedule:
+        async with AsyncClient(
+            transport=ASGITransport(app=test_app), base_url="http://test"
+        ) as client:
+            response = await client.delete(f"/tracks/comments/{comment_id}")
+
+    assert response.status_code == 200
+    assert response.json()["deleted"] is True
+
+    # no PDS deletion should be scheduled since there's no ATProto record
+    mock_schedule.assert_not_called()
+
+    # verify DB entry is still gone
     result = await db_session.execute(
         select(TrackComment).where(TrackComment.id == comment_id)
     )

backend/tests/api/test_track_likes.py

···
 """tests for track like api endpoints and error handling."""
 
 from collections.abc import Generator
-from unittest.mock import AsyncMock, patch
+from unittest.mock import patch
 
 import pytest
 from fastapi import FastAPI
···
 async def test_like_track_success(
     test_app: FastAPI, db_session: AsyncSession, test_track: Track
 ):
-    """test successful track like creates ATProto record and DB entry."""
-    with patch("backend.api.tracks.likes.create_like_record") as mock_create:
-        mock_create.return_value = "at://did:test:user123/fm.plyr.like/abc123"
-
+    """test successful track like creates DB entry and schedules PDS record creation."""
+    with patch("backend.api.tracks.likes.schedule_pds_create_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert response.status_code == 200
     assert response.json()["liked"] is True
 
-    # verify ATProto record was created
-    mock_create.assert_called_once()
+    # verify background task was scheduled
+    mock_schedule.assert_called_once()
+    call_kwargs = mock_schedule.call_args.kwargs
+    assert call_kwargs["subject_uri"] == test_track.atproto_record_uri
+    assert call_kwargs["subject_cid"] == test_track.atproto_record_cid
 
-    # verify DB entry exists
+    # verify DB entry exists (created immediately, before PDS)
     result = await db_session.execute(
         select(TrackLike).where(
             TrackLike.track_id == test_track.id,
···
     )
     like = result.scalar_one_or_none()
     assert like is not None
-    assert like.atproto_like_uri == "at://did:test:user123/fm.plyr.like/abc123"
+    # atproto_like_uri is None initially - will be set by background task
+    assert like.atproto_like_uri is None
 
 
-async def test_like_track_cleanup_on_db_failure(
+async def test_like_track_db_entry_has_correct_like_id(
     test_app: FastAPI, db_session: AsyncSession, test_track: Track
 ):
-    """test that ATProto record is cleaned up if DB commit fails."""
-    created_uri = "at://did:test:user123/fm.plyr.like/abc123"
-
-    # create a mock session that fails on commit
-    mock_db = AsyncMock(spec=AsyncSession)
-    mock_db.execute = db_session.execute
-    mock_db.add = db_session.add
-    mock_db.delete = db_session.delete
-    mock_db.flush = db_session.flush
-    mock_db.refresh = db_session.refresh
-    mock_db.commit = AsyncMock(side_effect=Exception("DB error"))
-
-    async def mock_get_db():
-        yield mock_db
-
-    from backend.models import get_db
-
-    test_app.dependency_overrides[get_db] = mock_get_db
-
-    with (
-        patch("backend.api.tracks.likes.create_like_record") as mock_create,
-        patch("backend.api.tracks.likes.delete_record_by_uri") as mock_delete,
-    ):
-        mock_create.return_value = created_uri
-
+    """test that the like_id passed to background task matches the DB record."""
+    with patch("backend.api.tracks.likes.schedule_pds_create_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
-            response = await client.post(f"/tracks/{test_track.id}/like")
+            await client.post(f"/tracks/{test_track.id}/like")
 
-    # should return 500 error
-    assert response.status_code == 500
-    assert "failed to like track" in response.json()["detail"]
+    # get the like_id from the scheduled call
+    call_kwargs = mock_schedule.call_args.kwargs
+    scheduled_like_id = call_kwargs["like_id"]
 
-    # verify cleanup was attempted
-    mock_delete.assert_called_once()
-    call_args = mock_delete.call_args
-    assert call_args.kwargs["record_uri"] == created_uri
-
-    # cleanup override
-    del test_app.dependency_overrides[get_db]
+    # verify it matches the DB record
+    result = await db_session.execute(
+        select(TrackLike).where(
+            TrackLike.track_id == test_track.id,
+            TrackLike.user_did == "did:test:user123",
+        )
+    )
+    like = result.scalar_one()
+    assert like.id == scheduled_like_id
 
 
 async def test_unlike_track_success(
     test_app: FastAPI, db_session: AsyncSession, test_track: Track
 ):
-    """test successful track unlike deletes ATProto record and DB entry."""
+    """test successful track unlike removes DB entry and schedules PDS record deletion."""
     # create existing like
     like = TrackLike(
         track_id=test_track.id,
···
     db_session.add(like)
     await db_session.commit()
 
-    with patch("backend.api.tracks.likes.delete_record_by_uri") as mock_delete:
+    with patch("backend.api.tracks.likes.schedule_pds_delete_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert response.status_code == 200
     assert response.json()["liked"] is False
 
-    # verify ATProto record was deleted
-    mock_delete.assert_called_once()
+    # verify background task was scheduled with correct URI
+    mock_schedule.assert_called_once()
+    call_kwargs = mock_schedule.call_args.kwargs
+    assert call_kwargs["like_uri"] == "at://did:test:user123/fm.plyr.like/abc123"
 
-    # verify DB entry is gone
+    # verify DB entry is gone (deleted immediately, before PDS)
     result = await db_session.execute(
         select(TrackLike).where(
             TrackLike.track_id == test_track.id,
···
     assert result.scalar_one_or_none() is None
 
 
-async def test_unlike_track_rollback_on_db_failure(
+async def test_unlike_track_without_atproto_uri(
     test_app: FastAPI, db_session: AsyncSession, test_track: Track
 ):
-    """test that ATProto record is recreated if DB commit fails during unlike."""
-    like_uri = "at://did:test:user123/fm.plyr.like/abc123"
-
-    # create existing like
+    """test that unliking a track without ATProto URI doesn't schedule deletion."""
+    # create like without ATProto URI (e.g., background task hasn't run yet)
     like = TrackLike(
         track_id=test_track.id,
         user_did="did:test:user123",
-        atproto_like_uri=like_uri,
+        atproto_like_uri=None,
     )
     db_session.add(like)
     await db_session.commit()
 
-    # create a mock session that fails on commit
-    mock_db = AsyncMock(spec=AsyncSession)
-    mock_db.execute = db_session.execute
-    mock_db.add = db_session.add
-    mock_db.delete = db_session.delete
-    mock_db.flush = db_session.flush
-    mock_db.refresh = db_session.refresh
-    mock_db.commit = AsyncMock(side_effect=Exception("DB error"))
-
-    async def mock_get_db():
-        yield mock_db
-
-    from backend.models import get_db
-
-    test_app.dependency_overrides[get_db] = mock_get_db
-
-    with (
-        patch("backend.api.tracks.likes.delete_record_by_uri") as mock_delete,
-        patch("backend.api.tracks.likes.create_like_record") as mock_create,
-    ):
-        mock_create.return_value = "at://did:test:user123/fm.plyr.like/new123"
-
+    with patch("backend.api.tracks.likes.schedule_pds_delete_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
             response = await client.delete(f"/tracks/{test_track.id}/like")
 
-    # should return 500 error
-    assert response.status_code == 500
-    assert "failed to unlike track" in response.json()["detail"]
+    assert response.status_code == 200
+    assert response.json()["liked"] is False
 
-    # verify ATProto record was deleted
-    mock_delete.assert_called_once()
+    # no PDS deletion should be scheduled since there's no ATProto record
+    mock_schedule.assert_not_called()
 
-    # verify rollback was attempted (recreate the like)
-    mock_create.assert_called_once()
-    call_args = mock_create.call_args
-    assert call_args.kwargs["subject_uri"] == test_track.atproto_record_uri
-    assert call_args.kwargs["subject_cid"] == test_track.atproto_record_cid
-
-    # cleanup override
-    del test_app.dependency_overrides[get_db]
+    # verify DB entry is still gone
+    result = await db_session.execute(
+        select(TrackLike).where(
+            TrackLike.track_id == test_track.id,
+            TrackLike.user_did == "did:test:user123",
+        )
+    )
+    assert result.scalar_one_or_none() is None
 
 
 async def test_like_already_liked_track_idempotent(
···
     db_session.add(like)
     await db_session.commit()
 
-    with patch("backend.api.tracks.likes.create_like_record") as mock_create:
+    with patch("backend.api.tracks.likes.schedule_pds_create_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert response.status_code == 200
     assert response.json()["liked"] is True
 
-    # verify no new ATProto record was created
-    mock_create.assert_not_called()
+    # verify no new background task was scheduled
+    mock_schedule.assert_not_called()
 
 
 async def test_unlike_not_liked_track_idempotent(
     test_app: FastAPI, db_session: AsyncSession, test_track: Track
 ):
     """test that unliking a not-liked track is idempotent."""
-    with patch("backend.api.tracks.likes.delete_record_by_uri") as mock_delete:
+    with patch("backend.api.tracks.likes.schedule_pds_delete_like") as mock_schedule:
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
         ) as client:
···
     assert response.status_code == 200
     assert response.json()["liked"] is False
 
-    # verify no ATProto record deletion was attempted
-    mock_delete.assert_not_called()
+    # verify no background task was scheduled
+    mock_schedule.assert_not_called()
+
+
+async def test_like_track_missing_atproto_record(
+    test_app: FastAPI, db_session: AsyncSession
+):
+    """test that liking a track without ATProto record returns 422."""
+    # create artist
+    artist = Artist(
+        did="did:plc:artist456",
+        handle="artist2.bsky.social",
+        display_name="Test Artist 2",
+    )
+    db_session.add(artist)
+    await db_session.flush()
+
+    # create track WITHOUT ATProto record
+    track = Track(
+        title="No ATProto Track",
+        artist_did=artist.did,
+        file_id="noatproto123",
+        file_type="mp3",
+        atproto_record_uri=None,
+        atproto_record_cid=None,
+    )
+    db_session.add(track)
+    await db_session.commit()
+    await db_session.refresh(track)
+
+    async with AsyncClient(
+        transport=ASGITransport(app=test_app), base_url="http://test"
+    ) as client:
+        response = await client.post(f"/tracks/{track.id}/like")
+
+    assert response.status_code == 422
+    detail = response.json()["detail"]
+    assert detail["error"] == "missing_atproto_record"
+
+
+async def test_like_nonexistent_track(test_app: FastAPI):
+    """test that liking a nonexistent track returns 404."""
+    async with AsyncClient(
+        transport=ASGITransport(app=test_app), base_url="http://test"
+    ) as client:
+        response = await client.post("/tracks/99999/like")
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "track not found"