feat: migrate ATProto sync and teal scrobbling to docket (#539)

* feat: migrate ATProto sync and teal scrobbling to docket

- add sync_atproto and scrobble_to_teal as docket background tasks
- remove all fallback/bifurcation code - Redis is always required
- simplify background.py (remove is_docket_enabled)
- update auth.py and playback.py to use new schedulers
- add Redis service to CI workflow
- update tests for simplified docket-only flow

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

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

* fix: update test for schedule_atproto_sync, fix cookies deprecation

- test_list_record_sync: patch schedule_atproto_sync instead of removed sync_atproto_records
- test_hidden_tags_filter: move cookies to client constructor to fix httpx deprecation warning

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

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

* feat: template database pattern for fast parallel test execution

- use template db + clone for xdist workers (instant file copy vs migrations)
- advisory locks coordinate template creation between workers
- patch settings.database.url per-worker for production code compatibility
- borrowed pattern from prefecthq/nebula

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

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

* ci: enable parallel test execution with xdist

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

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

* fix: ensure test_app fixture depends on db_session for xdist

fixes race condition where tests using test_app without db_session
would run before database URL was patched for the worker

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

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

---------

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

.github/workflows/test-backend.yml

···
           --health-timeout 5s
           --health-retries 5
 
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     steps:
       - uses: actions/checkout@v5
 
···
       - name: run tests
         env:
           DATABASE_URL: postgresql+asyncpg://relay_test:relay_test@localhost:5432/relay_test
-        run: cd backend && uv run pytest tests/
+          DOCKET_URL: redis://localhost:6379
+        run: cd backend && uv run pytest tests/ -n auto
 
       - name: prune uv cache
         if: always()

backend/src/backend/_internal/background.py

···
 
 provides a docket instance for scheduling background tasks and a worker
 that runs alongside the FastAPI server. requires DOCKET_URL to be set
-to a Redis URL for durable execution across multiple machines.
+to a Redis URL.
 
 usage:
-    from backend._internal.background import get_docket, is_docket_enabled
+    from backend._internal.background import get_docket
 
-    if is_docket_enabled():
-        docket = get_docket()
-        await docket.add(my_task_function)(arg1, arg2)
-    else:
-        # fallback to direct execution or FastAPI BackgroundTasks
-        await my_task_function(arg1, arg2)
+    docket = get_docket()
+    await docket.add(my_task_function)(arg1, arg2)
 """
 
 import asyncio
···
 
 logger = logging.getLogger(__name__)
 
-# global docket instance - initialized in lifespan (None if disabled)
+# global docket instance - initialized in lifespan
 _docket: Docket | None = None
-_docket_enabled: bool = False
-
-
-def is_docket_enabled() -> bool:
-    """check if docket is enabled and initialized."""
-    return _docket_enabled and _docket is not None
 
 
 def get_docket() -> Docket:
     """get the global docket instance.
 
     raises:
-        RuntimeError: if docket is not initialized or disabled
+        RuntimeError: if docket is not initialized
     """
-    if not _docket_enabled:
-        raise RuntimeError("docket is disabled - set DOCKET_URL to enable")
     if _docket is None:
         raise RuntimeError("docket not initialized - is the server running?")
     return _docket
 
 
 @asynccontextmanager
-async def background_worker_lifespan() -> AsyncGenerator[Docket | None, None]:
+async def background_worker_lifespan() -> AsyncGenerator[Docket, None]:
     """lifespan context manager for docket and its worker.
 
-    if DOCKET_URL is not set, docket is disabled and this yields None.
-    when enabled, initializes the docket connection and starts an in-process
+    initializes the docket connection and starts an in-process
     worker that processes background tasks.
 
     yields:
-        Docket | None: the initialized docket instance, or None if disabled
+        Docket: the initialized docket instance
     """
-    global _docket, _docket_enabled
-
-    # check if docket should be enabled
-    if not settings.docket.url:
-        logger.info("docket disabled (DOCKET_URL not set)")
-        _docket_enabled = False
-        yield None
-        return
+    global _docket
 
-    _docket_enabled = True
     logger.info(
         "initializing docket",
         extra={"docket_name": settings.docket.name, "url": settings.docket.url},
···
                     await worker_task
                 except asyncio.CancelledError:
                     logger.debug("docket worker task cancelled")
-            # clear globals after worker is fully stopped
+            # clear global after worker is fully stopped
             _docket = None
-            _docket_enabled = False
             logger.info("docket worker stopped")
 
 
···
     add new task imports here as they're created.
     """
     # import task functions here to avoid circular imports
-    from backend._internal.background_tasks import process_export, scan_copyright
+    from backend._internal.background_tasks import (
+        process_export,
+        scan_copyright,
+        scrobble_to_teal,
+        sync_atproto,
+    )
 
     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"]},
+        extra={
+            "tasks": [
+                "scan_copyright",
+                "process_export",
+                "sync_atproto",
+                "scrobble_to_teal",
+            ]
+        },
     )

backend/src/backend/_internal/background_tasks.py

···
 
 these functions are registered with docket and executed by workers.
 they should be self-contained and handle their own database sessions.
+
+requires DOCKET_URL to be set (Redis is always available).
 """
 
-import asyncio
 import logging
 import os
 import tempfile
···
 import aiofiles
 import logfire
 
-from backend._internal.background import get_docket, is_docket_enabled
+from backend._internal.background import get_docket
 
 logger = logging.getLogger(__name__)
 
 
 async def scan_copyright(track_id: int, audio_url: str) -> None:
     """scan a track for potential copyright matches.
-
-    this is the docket version of the copyright scan task. when docket
-    is enabled (DOCKET_URL set), this provides durability and retries
-    compared to fire-and-forget asyncio.create_task().
 
     args:
         track_id: database ID of the track to scan
···
 
 
 async def schedule_copyright_scan(track_id: int, audio_url: str) -> None:
-    """schedule a copyright scan, using docket if enabled, else asyncio.
-
-    this is the entry point for scheduling copyright scans. it handles
-    the docket vs asyncio fallback logic in one place.
-    """
-    from backend._internal.moderation import scan_track_for_copyright
-
-    if is_docket_enabled():
-        try:
-            docket = get_docket()
-            await docket.add(scan_copyright)(track_id, audio_url)
-            logfire.info("scheduled copyright scan via docket", track_id=track_id)
-            return
-        except Exception as e:
-            logfire.warning(
-                "docket scheduling failed, falling back to asyncio",
-                track_id=track_id,
-                error=str(e),
-            )
-
-    # fallback: fire-and-forget
-    asyncio.create_task(scan_track_for_copyright(track_id, audio_url))  # noqa: RUF006
+    """schedule a copyright scan via docket."""
+    docket = get_docket()
+    await docket.add(scan_copyright)(track_id, audio_url)
+    logfire.info("scheduled copyright scan", track_id=track_id)
 
 
 async def process_export(export_id: str, artist_did: str) -> None:
···
 
 
 async def schedule_export(export_id: str, artist_did: str) -> None:
-    """schedule an export, using docket if enabled, else asyncio.
+    """schedule an export via docket."""
+    docket = get_docket()
+    await docket.add(process_export)(export_id, artist_did)
+    logfire.info("scheduled export", export_id=export_id)
+
+
+async def sync_atproto(session_id: str, user_did: str) -> None:
+    """sync ATProto records (profile, albums, liked tracks) for a user.
+
+    this runs after login or scope upgrade to ensure the user's PDS
+    has up-to-date records for their plyr.fm data.
+
+    args:
+        session_id: the user's session ID for authentication
+        user_did: the user's DID
+    """
+    from backend._internal.atproto.sync import sync_atproto_records
+    from backend._internal.auth import get_session
+
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"sync_atproto: session {session_id[:8]}... not found")
+        return
+
+    await sync_atproto_records(auth_session, user_did)
+
+
+async def schedule_atproto_sync(session_id: str, user_did: str) -> None:
+    """schedule an ATProto sync via docket."""
+    docket = get_docket()
+    await docket.add(sync_atproto)(session_id, user_did)
+    logfire.info("scheduled atproto sync", user_did=user_did)
+
+
+async def scrobble_to_teal(
+    session_id: str,
+    track_id: int,
+    track_title: str,
+    artist_name: str,
+    duration: int | None,
+    album_name: str | None,
+) -> None:
+    """scrobble a play to teal.fm (creates play record + updates status).
 
-    this is the entry point for scheduling exports. it handles
-    the docket vs asyncio fallback logic in one place.
+    args:
+        session_id: the user's session ID for authentication
+        track_id: database ID of the track
+        track_title: title of the track
+        artist_name: name of the artist
+        duration: track duration in seconds
+        album_name: album name (optional)
     """
-    if is_docket_enabled():
-        try:
-            docket = get_docket()
-            await docket.add(process_export)(export_id, artist_did)
-            logfire.info("scheduled export via docket", export_id=export_id)
-            return
-        except Exception as e:
-            logfire.warning(
-                "docket scheduling failed, falling back to asyncio",
-                export_id=export_id,
-                error=str(e),
-            )
+    from backend._internal.atproto.teal import (
+        create_teal_play_record,
+        update_teal_status,
+    )
+    from backend._internal.auth import get_session
+    from backend.config import settings
+
+    auth_session = await get_session(session_id)
+    if not auth_session:
+        logger.warning(f"teal scrobble: session {session_id[:8]}... not found")
+        return
 
-    # fallback: fire-and-forget
-    asyncio.create_task(process_export(export_id, artist_did))  # noqa: RUF006
+    origin_url = f"{settings.frontend.url}/track/{track_id}"
+
+    try:
+        # create play record (scrobble)
+        play_uri = await create_teal_play_record(
+            auth_session=auth_session,
+            track_name=track_title,
+            artist_name=artist_name,
+            duration=duration,
+            album_name=album_name,
+            origin_url=origin_url,
+        )
+        logger.info(f"teal play record created: {play_uri}")
+
+        # update status (now playing)
+        status_uri = await update_teal_status(
+            auth_session=auth_session,
+            track_name=track_title,
+            artist_name=artist_name,
+            duration=duration,
+            album_name=album_name,
+            origin_url=origin_url,
+        )
+        logger.info(f"teal status updated: {status_uri}")
+
+    except Exception as e:
+        logger.error(f"teal scrobble failed for track {track_id}: {e}", exc_info=True)
+
+
+async def schedule_teal_scrobble(
+    session_id: str,
+    track_id: int,
+    track_title: str,
+    artist_name: str,
+    duration: int | None,
+    album_name: str | None,
+) -> None:
+    """schedule a teal scrobble via docket."""
+    docket = get_docket()
+    await docket.add(scrobble_to_teal)(
+        session_id, track_id, track_title, artist_name, duration, album_name
+    )
+    logfire.info("scheduled teal scrobble", track_id=track_id)

backend/src/backend/api/auth.py

···
 """authentication api endpoints."""
 
-import asyncio
 import logging
 from typing import Annotated
 
···
     start_oauth_flow,
     start_oauth_flow_with_scopes,
 )
-from backend._internal.atproto import sync_atproto_records
+from backend._internal.background_tasks import schedule_atproto_sync
 from backend.config import settings
 from backend.utilities.rate_limit import limiter
 
 logger = logging.getLogger(__name__)
-
-# hold references to background tasks to prevent GC before completion
-_background_tasks: set[asyncio.Task[None]] = set()
-
-
-def _create_background_task(coro) -> asyncio.Task:
-    """create a background task with proper lifecycle management."""
-    task = asyncio.create_task(coro)
-    _background_tasks.add(task)
-    task.add_done_callback(_background_tasks.discard)
-    return task
-
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
···
         # create exchange token - NOT marked as dev token so cookie gets set
         exchange_token = await create_exchange_token(session_id)
 
-        # fire-and-forget: sync ATProto records with new scopes
-        auth_session = Session(
-            session_id=session_id,
-            did=did,
-            handle=handle,
-            oauth_session=oauth_session,
-        )
-
-        async def _sync_on_scope_upgrade():
-            try:
-                await sync_atproto_records(auth_session, did)
-            except Exception as e:
-                logger.error(f"background sync failed for {did}: {e}", exc_info=True)
-
-        _create_background_task(_sync_on_scope_upgrade())
+        # schedule ATProto sync (via docket if enabled, else asyncio)
+        await schedule_atproto_sync(session_id, did)
 
         return RedirectResponse(
             url=f"{settings.frontend.url}/settings?exchange_token={exchange_token}&scope_upgraded=true",
···
     # check if artist profile exists
     has_profile = await check_artist_profile_exists(did)
 
-    # fire-and-forget: sync ATProto records on login
-    auth_session = Session(
-        session_id=session_id,
-        did=did,
-        handle=handle,
-        oauth_session=oauth_session,
-    )
-
-    async def _sync_on_login():
-        try:
-            await sync_atproto_records(auth_session, did)
-        except Exception as e:
-            logger.error(f"background sync failed for {did}: {e}", exc_info=True)
-
-    _create_background_task(_sync_on_login())
+    # schedule ATProto sync (via docket if enabled, else asyncio)
+    await schedule_atproto_sync(session_id, did)
 
     # redirect to profile setup if needed, otherwise to portal
     redirect_path = "/portal" if has_profile else "/profile/setup"

backend/src/backend/api/tracks/playback.py

···
 import logging
 from typing import Annotated
 
-from fastapi import BackgroundTasks, Depends, HTTPException
+from fastapi import Depends, HTTPException
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from backend._internal import Session, get_optional_session
-from backend._internal.atproto.teal import create_teal_play_record, update_teal_status
-from backend._internal.auth import get_session
+from backend._internal.background_tasks import schedule_teal_scrobble
 from backend.config import settings
 from backend.models import Artist, Track, TrackLike, UserPreferences, get_db
 from backend.schemas import TrackResponse
···
     )
 
 
-async def _scrobble_to_teal(
-    session_id: str,
-    track_id: int,
-    track_title: str,
-    artist_name: str,
-    duration: int | None,
-    album_name: str | None,
-) -> None:
-    """scrobble a play to teal.fm (creates play record + updates status)."""
-    if not (auth_session := await get_session(session_id)):
-        logger.warning(f"teal scrobble failed: session {session_id[:8]}... not found")
-        return
-
-    origin_url = f"{settings.frontend.url}/track/{track_id}"
-
-    try:
-        # create play record (scrobble)
-        play_uri = await create_teal_play_record(
-            auth_session=auth_session,
-            track_name=track_title,
-            artist_name=artist_name,
-            duration=duration,
-            album_name=album_name,
-            origin_url=origin_url,
-        )
-        logger.info(f"teal play record created: {play_uri}")
-
-        # update status (now playing)
-        status_uri = await update_teal_status(
-            auth_session=auth_session,
-            track_name=track_title,
-            artist_name=artist_name,
-            duration=duration,
-            album_name=album_name,
-            origin_url=origin_url,
-        )
-        logger.info(f"teal status updated: {status_uri}")
-
-    except Exception as e:
-        logger.error(f"teal scrobble failed for track {track_id}: {e}", exc_info=True)
-
-
 @router.post("/{track_id}/play")
 async def increment_play_count(
     track_id: int,
     db: Annotated[AsyncSession, Depends(get_db)],
-    background_tasks: BackgroundTasks,
     session: Session | None = Depends(get_optional_session),
 ) -> dict:
     """Increment play count for a track (called after 30 seconds of playback).
···
             # check if session has teal scopes
             scope = session.oauth_session.get("scope", "")
             if settings.teal.play_collection in scope:
-                background_tasks.add_task(
-                    _scrobble_to_teal,
+                await schedule_teal_scrobble(
                     session_id=session.session_id,
                     track_id=track_id,
                     track_title=track.title,

backend/tests/api/test_hidden_tags_filter.py

···
 ):
     """test that discovery feed (no artist_did) filters hidden tags."""
     async with AsyncClient(
-        transport=ASGITransport(app=test_app), base_url="http://test"
+        transport=ASGITransport(app=test_app),
+        base_url="http://test",
+        cookies={"session_id": "test_session"},
     ) as client:
-        response = await client.get(
-            "/tracks/",
-            cookies={"session_id": "test_session"},
-        )
+        response = await client.get("/tracks/")
 
     assert response.status_code == 200
     tracks = response.json()["tracks"]
···
 ):
     """test that artist page (with artist_did) shows all tracks including hidden."""
     async with AsyncClient(
-        transport=ASGITransport(app=test_app), base_url="http://test"
+        transport=ASGITransport(app=test_app),
+        base_url="http://test",
+        cookies={"session_id": "test_session"},
     ) as client:
-        response = await client.get(
-            f"/tracks/?artist_did={artist.did}",
-            cookies={"session_id": "test_session"},
-        )
+        response = await client.get(f"/tracks/?artist_did={artist.did}")
 
     assert response.status_code == 200
     tracks = response.json()["tracks"]
···
 ):
     """test that filter_hidden_tags=true forces filtering even on artist page."""
     async with AsyncClient(
-        transport=ASGITransport(app=test_app), base_url="http://test"
+        transport=ASGITransport(app=test_app),
+        base_url="http://test",
+        cookies={"session_id": "test_session"},
     ) as client:
         response = await client.get(
-            f"/tracks/?artist_did={artist.did}&filter_hidden_tags=true",
-            cookies={"session_id": "test_session"},
+            f"/tracks/?artist_did={artist.did}&filter_hidden_tags=true"
         )
 
     assert response.status_code == 200
···
 ):
     """test that filter_hidden_tags=false disables filtering on discovery feed."""
     async with AsyncClient(
-        transport=ASGITransport(app=test_app), base_url="http://test"
+        transport=ASGITransport(app=test_app),
+        base_url="http://test",
+        cookies={"session_id": "test_session"},
     ) as client:
-        response = await client.get(
-            "/tracks/?filter_hidden_tags=false",
-            cookies={"session_id": "test_session"},
-        )
+        response = await client.get("/tracks/?filter_hidden_tags=false")
 
     assert response.status_code == 200
     tracks = response.json()["tracks"]

backend/tests/api/test_list_record_sync.py

···
 """tests for ATProto list record sync on login."""
 
-import asyncio
 from collections.abc import Generator
 from unittest.mock import AsyncMock, patch
 
···
             return_value=True,
         ),
         patch(
-            "backend.api.auth.sync_atproto_records",
+            "backend.api.auth.schedule_atproto_sync",
             new_callable=AsyncMock,
-        ) as mock_sync,
+        ) as mock_schedule_sync,
     ):
         async with AsyncClient(
             transport=ASGITransport(app=test_app), base_url="http://test"
···
                 follow_redirects=False,
             )
 
-        # give background tasks time to run
-        await asyncio.sleep(0.1)
-
     assert response.status_code == 303
     assert "exchange_token=test_exchange_token" in response.headers["location"]
 
-    # verify sync was triggered in background
-    mock_sync.assert_called_once()
-    call_args = mock_sync.call_args
-    # first arg is the session, second is the DID
-    assert call_args[0][1] == "did:plc:testartist123"
+    # verify sync was scheduled via docket
+    mock_schedule_sync.assert_called_once_with(
+        "test_session_id", "did:plc:testartist123"
+    )

backend/tests/api/test_oembed.py

···
 
 
 @pytest.fixture
-def test_app() -> FastAPI:
-    """get test app."""
+def test_app(db_session: AsyncSession) -> FastAPI:
+    """get test app with db session dependency to ensure correct database URL."""
+    _ = db_session  # ensures database fixtures run first
     return app
 
 

backend/tests/conftest.py

···
 from collections.abc import AsyncGenerator, Generator
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
+from urllib.parse import urlsplit, urlunsplit
 
+import asyncpg
 import pytest
 import sqlalchemy as sa
 from fastapi import FastAPI
···
 
     # set _storage directly to prevent R2Storage initialization
     backend.storage._storage = MockStorage()  # type: ignore[assignment]
+
+
+def _database_from_url(url: str) -> str:
+    """extract database name from connection URL."""
+    _, _, path, _, _ = urlsplit(url)
+    return path.strip("/")
+
+
+def _postgres_admin_url(database_url: str) -> str:
+    """convert async database URL to sync postgres URL for admin operations."""
+    scheme, netloc, _, query, fragment = urlsplit(database_url)
+    # asyncpg -> postgres for direct connection
+    scheme = scheme.replace("+asyncpg", "").replace("postgresql", "postgres")
+    return urlunsplit((scheme, netloc, "/postgres", query, fragment))
 
 
 @asynccontextmanager
···
     await connection.execute(sa.text(stmt))
 
 
+async def _setup_template_database(template_url: str) -> None:
+    """initialize database schema and helper procedure on template database."""
+    engine = create_async_engine(template_url, echo=False)
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            await _truncate_tables(conn)
+            await _create_clear_database_procedure(conn)
+    finally:
+        await engine.dispose()
+
+
+async def _ensure_template_database(base_url: str) -> str:
+    """ensure template database exists and is migrated.
+
+    uses advisory lock to coordinate between xdist workers.
+    returns the template database name.
+    """
+    base_db_name = _database_from_url(base_url)
+    template_db_name = f"{base_db_name}_template"
+    postgres_url = _postgres_admin_url(base_url)
+
+    conn = await asyncpg.connect(postgres_url)
+    try:
+        # advisory lock prevents race condition between workers
+        await conn.execute("SELECT pg_advisory_lock(hashtext($1))", template_db_name)
+
+        # check if template exists
+        exists = await conn.fetchval(
+            "SELECT 1 FROM pg_database WHERE datname = $1", template_db_name
+        )
+
+        if not exists:
+            # create template database
+            await conn.execute(f'CREATE DATABASE "{template_db_name}"')
+
+            # build URL for template and set it up
+            scheme, netloc, _, query, fragment = urlsplit(base_url)
+            template_url = urlunsplit(
+                (scheme, netloc, f"/{template_db_name}", query, fragment)
+            )
+            await _setup_template_database(template_url)
+
+        # release lock (other workers waiting will see template exists)
+        await conn.execute("SELECT pg_advisory_unlock(hashtext($1))", template_db_name)
+
+        return template_db_name
+    finally:
+        await conn.close()
+
+
+async def _create_worker_database_from_template(
+    base_url: str, worker_id: str, template_db_name: str
+) -> str:
+    """create worker database by cloning the template (instant file copy)."""
+    base_db_name = _database_from_url(base_url)
+    worker_db_name = f"{base_db_name}_{worker_id}"
+    postgres_url = _postgres_admin_url(base_url)
+
+    conn = await asyncpg.connect(postgres_url)
+    try:
+        # kill connections to worker db (if it exists from previous run)
+        await conn.execute(
+            """
+            SELECT pg_terminate_backend(pid)
+            FROM pg_stat_activity
+            WHERE datname = $1 AND pid <> pg_backend_pid()
+            """,
+            worker_db_name,
+        )
+
+        # kill connections to template db (required for cloning)
+        await conn.execute(
+            """
+            SELECT pg_terminate_backend(pid)
+            FROM pg_stat_activity
+            WHERE datname = $1 AND pid <> pg_backend_pid()
+            """,
+            template_db_name,
+        )
+
+        # drop and recreate from template (instant - just file copy)
+        await conn.execute(f'DROP DATABASE IF EXISTS "{worker_db_name}"')
+        await conn.execute(
+            f'CREATE DATABASE "{worker_db_name}" WITH TEMPLATE "{template_db_name}"'
+        )
+
+        return worker_db_name
+    finally:
+        await conn.close()
+
+
 @pytest.fixture(scope="session")
 def test_database_url(worker_id: str) -> str:
     """generate a unique test database URL for each pytest worker.
 
-    reads from settings.database.url and appends worker suffix if needed.
+    uses template database pattern for fast parallel test execution:
+    1. first worker creates template db with migrations (once)
+    2. each worker clones from template (instant file copy)
+
+    also patches settings.database.url so all production code uses test db.
     """
+    import asyncio
+    import os
+
     base_url = settings.database.url
 
-    # for parallel test execution, append worker id to database name
+    # single worker - just use base database
     if worker_id == "master":
+        asyncio.run(_setup_database_direct(base_url))
         return base_url
 
-    # worker_id will be "gw0", "gw1", etc for xdist workers
-    return f"{base_url}_{worker_id}"
+    # xdist workers - use template pattern
+    template_db_name = asyncio.run(_ensure_template_database(base_url))
+    asyncio.run(
+        _create_worker_database_from_template(base_url, worker_id, template_db_name)
+    )
 
+    # build URL for worker database
+    scheme, netloc, _, query, fragment = urlsplit(base_url)
+    base_db_name = _database_from_url(base_url)
+    worker_db_name = f"{base_db_name}_{worker_id}"
+    worker_url = urlunsplit((scheme, netloc, f"/{worker_db_name}", query, fragment))
 
-async def _setup_database(test_database_url: str) -> None:
-    """initialize database schema and helper procedure."""
-    engine = create_async_engine(test_database_url, echo=False)
+    # patch settings so all production code uses this URL
+    # this is safe because each xdist worker is a separate process
+    settings.database.url = worker_url
+    os.environ["DATABASE_URL"] = worker_url
+
+    return worker_url
+
+
+async def _setup_database_direct(database_url: str) -> None:
+    """set up database directly (for single worker mode)."""
+    engine = create_async_engine(database_url, echo=False)
     try:
         async with engine.begin() as conn:
             await conn.run_sync(Base.metadata.create_all)
···
 
 @pytest.fixture(scope="session")
 def _database_setup(test_database_url: str) -> None:
-    """create tables and stored procedures once per test session."""
-    import asyncio
-
-    asyncio.run(_setup_database(test_database_url))
+    """marker fixture - database is set up by test_database_url fixture."""
+    _ = test_database_url  # ensure dependency chain
 
 
 @pytest.fixture()
···
     test_database_url: str, _database_setup: None
 ) -> AsyncGenerator[AsyncEngine, None]:
     """create a database engine for each test (to avoid event loop issues)."""
+    from backend.utilities.database import ENGINES
+
+    # clear any cached engines from previous tests
+    for cached_engine in list(ENGINES.values()):
+        await cached_engine.dispose()
+    ENGINES.clear()
+
     engine = create_async_engine(
         test_database_url,
         echo=False,
···
         yield engine
     finally:
         await engine.dispose()
-        # also dispose any engines cached by production code (database.py)
-        # to prevent connection accumulation across tests
-        from backend.utilities.database import ENGINES
-
+        # clean up cached engines
         for cached_engine in list(ENGINES.values()):
             await cached_engine.dispose()
         ENGINES.clear()

backend/tests/test_background_tasks.py

···
 import backend._internal.background_tasks as bg_tasks
 
 
-async def test_schedule_export_uses_docket_when_enabled() -> None:
-    """when docket is enabled, schedule_export should add task to docket."""
+async def test_schedule_export_uses_docket() -> None:
+    """schedule_export should add task to docket."""
     calls: list[tuple[str, str]] = []
 
     async def mock_schedule(export_id: str, artist_did: str) -> None:
···
     mock_docket.add = MagicMock(return_value=mock_schedule)
 
     with (
-        patch.object(bg_tasks, "is_docket_enabled", return_value=True),
         patch.object(bg_tasks, "get_docket", return_value=mock_docket),
         patch.object(bg_tasks, "process_export", MagicMock()),
     ):
···
         assert calls == [("export-123", "did:plc:testuser")]
 
 
-async def test_schedule_export_falls_back_to_asyncio_when_disabled() -> None:
-    """when docket is disabled, schedule_export should use asyncio.create_task."""
-    create_task_calls: list[object] = []
+async def test_schedule_copyright_scan_uses_docket() -> None:
+    """schedule_copyright_scan should add task to docket."""
+    calls: list[tuple[int, str]] = []
 
-    def capture_create_task(coro: object) -> MagicMock:
-        create_task_calls.append(coro)
-        return MagicMock()
+    async def mock_schedule(track_id: int, audio_url: str) -> None:
+        calls.append((track_id, audio_url))
 
-    process_export_calls: list[tuple[str, str]] = []
-
-    def mock_process_export(export_id: str, artist_did: str) -> object:
-        process_export_calls.append((export_id, artist_did))
-        return MagicMock()  # return non-coroutine to avoid unawaited warning
-
-    with (
-        patch.object(bg_tasks, "is_docket_enabled", return_value=False),
-        patch.object(bg_tasks, "process_export", mock_process_export),
-        patch.object(bg_tasks.asyncio, "create_task", capture_create_task),
-    ):
-        await bg_tasks.schedule_export("export-456", "did:plc:testuser")
-
-        assert len(create_task_calls) == 1
-        assert process_export_calls == [("export-456", "did:plc:testuser")]
-
-
-async def test_schedule_export_falls_back_on_docket_error() -> None:
-    """if docket scheduling fails, should fall back to asyncio."""
     mock_docket = MagicMock()
-    mock_docket.add.side_effect = Exception("redis connection failed")
-
-    create_task_calls: list[object] = []
-
-    def capture_create_task(coro: object) -> MagicMock:
-        create_task_calls.append(coro)
-        return MagicMock()
-
-    process_export_calls: list[tuple[str, str]] = []
-
-    def mock_process_export(export_id: str, artist_did: str) -> object:
-        process_export_calls.append((export_id, artist_did))
-        return MagicMock()
+    mock_docket.add = MagicMock(return_value=mock_schedule)
 
     with (
-        patch.object(bg_tasks, "is_docket_enabled", return_value=True),
         patch.object(bg_tasks, "get_docket", return_value=mock_docket),
-        patch.object(bg_tasks, "process_export", mock_process_export),
-        patch.object(bg_tasks.asyncio, "create_task", capture_create_task),
+        patch.object(bg_tasks, "scan_copyright", MagicMock()),
     ):
-        await bg_tasks.schedule_export("export-789", "did:plc:testuser")
+        await bg_tasks.schedule_copyright_scan(123, "https://example.com/audio.mp3")
 
-        assert len(create_task_calls) == 1
+        mock_docket.add.assert_called_once()
+        assert calls == [(123, "https://example.com/audio.mp3")]
 
 
-async def test_schedule_copyright_scan_uses_docket_when_enabled() -> None:
-    """when docket is enabled, schedule_copyright_scan should add task to docket."""
-    calls: list[tuple[int, str]] = []
+async def test_schedule_atproto_sync_uses_docket() -> None:
+    """schedule_atproto_sync should add task to docket."""
+    calls: list[tuple[str, str]] = []
 
-    async def mock_schedule(track_id: int, audio_url: str) -> None:
-        calls.append((track_id, audio_url))
+    async def mock_schedule(session_id: str, user_did: str) -> None:
+        calls.append((session_id, user_did))
 
     mock_docket = MagicMock()
     mock_docket.add = MagicMock(return_value=mock_schedule)
 
     with (
-        patch.object(bg_tasks, "is_docket_enabled", return_value=True),
         patch.object(bg_tasks, "get_docket", return_value=mock_docket),
-        patch.object(bg_tasks, "scan_copyright", MagicMock()),
+        patch.object(bg_tasks, "sync_atproto", MagicMock()),
     ):
-        await bg_tasks.schedule_copyright_scan(123, "https://example.com/audio.mp3")
+        await bg_tasks.schedule_atproto_sync("session-abc", "did:plc:testuser")
 
         mock_docket.add.assert_called_once()
-        assert calls == [(123, "https://example.com/audio.mp3")]
+        assert calls == [("session-abc", "did:plc:testuser")]
 
 
-async def test_schedule_copyright_scan_falls_back_to_asyncio_when_disabled() -> None:
-    """when docket is disabled, schedule_copyright_scan should use asyncio."""
-    create_task_calls: list[object] = []
+async def test_schedule_teal_scrobble_uses_docket() -> None:
+    """schedule_teal_scrobble should add task to docket."""
+    calls: list[tuple] = []
 
-    def capture_create_task(coro: object) -> MagicMock:
-        create_task_calls.append(coro)
-        return MagicMock()
+    async def mock_schedule(
+        session_id: str,
+        track_id: int,
+        track_title: str,
+        artist_name: str,
+        duration: int | None,
+        album_name: str | None,
+    ) -> None:
+        calls.append(
+            (session_id, track_id, track_title, artist_name, duration, album_name)
+        )
 
-    scan_calls: list[tuple[int, str]] = []
-
-    def mock_scan(track_id: int, audio_url: str) -> object:
-        scan_calls.append((track_id, audio_url))
-        return MagicMock()
+    mock_docket = MagicMock()
+    mock_docket.add = MagicMock(return_value=mock_schedule)
 
     with (
-        patch.object(bg_tasks, "is_docket_enabled", return_value=False),
-        patch("backend._internal.moderation.scan_track_for_copyright", mock_scan),
-        patch.object(bg_tasks.asyncio, "create_task", capture_create_task),
+        patch.object(bg_tasks, "get_docket", return_value=mock_docket),
+        patch.object(bg_tasks, "scrobble_to_teal", MagicMock()),
     ):
-        await bg_tasks.schedule_copyright_scan(456, "https://example.com/audio.mp3")
+        await bg_tasks.schedule_teal_scrobble(
+            session_id="session-xyz",
+            track_id=42,
+            track_title="Test Track",
+            artist_name="Test Artist",
+            duration=180,
+            album_name="Test Album",
+        )
 
-        assert len(create_task_calls) == 1
-        assert scan_calls == [(456, "https://example.com/audio.mp3")]
+        mock_docket.add.assert_called_once()
+        assert calls == [
+            ("session-xyz", 42, "Test Track", "Test Artist", 180, "Test Album")
+        ]