feat(api): offline-first sync layer

Implements the stubbed /sync endpoints:
- GET /sync/changes — delta pull (likes, plays, changed playlists with
  their track ids, changed catalogue tracks) over the half-open window
  (since, cursor]; the cursor is the DB clock, so it's immune to app/DB
  skew.
- POST /sync/push — idempotent append of client like/play events
  (ON CONFLICT DO NOTHING by client-supplied id); events for tracks the
  server doesn't have are skipped (graceful degradation).

Adds a server-ingestion column `synced_at` to the likes + play_history
event logs (migration) as the delta ordering key, so an event pushed
with an older event time still surfaces for other devices. Repos gain
list_since/add_event (likes, history), list_changed_since (playlists,
tracks) and playlist.track_ids; wired via SyncService in deps.

Also fixes test isolation exposed by the registry-backed /admin/sources
endpoint: test_sources_api clears the process-cached source registry,
and test_admin_api no longer hardcodes the environment name.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/20260728_1000-user_settings.py

@@ -0,0 +1,58 @@
+"""user_settings: per-user preferences + scrobbling config
+
+Revision ID: 20260728_user_settings
+Revises: dc126696f5a6
+Create Date: 2026-07-28 10:00:00.000000
+
+Adds the ``user_settings`` table (1:1 with ``users``, PK = user_id): general
+preferences (theme, stream quality) plus scrobbling config. The scrobbler
+session key is stored Fernet-encrypted, never in plaintext.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260728_user_settings"
+down_revision: str | None = "dc126696f5a6"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_settings",
+        sa.Column("user_id", sa.Uuid(), nullable=False),
+        sa.Column("theme", sa.String(length=16), nullable=False),
+        sa.Column("stream_quality", sa.String(length=16), nullable=False),
+        sa.Column("scrobble_enabled", sa.Boolean(), nullable=False),
+        sa.Column("scrobble_provider", sa.String(length=16), nullable=True),
+        sa.Column("scrobble_username", sa.String(length=255), nullable=True),
+        sa.Column("scrobble_session_key_enc", sa.String(length=512), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+            name=op.f("fk_user_settings_user_id_users"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("user_id", name=op.f("pk_user_settings")),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("user_settings")

alembic/versions/20260728_1100-sync_ingestion_columns.py

@@ -0,0 +1,57 @@
+"""sync: server-ingestion columns on the event logs
+
+Revision ID: 20260728_sync_synced_at
+Revises: 20260728_user_settings
+Create Date: 2026-07-28 11:00:00.000000
+
+Adds ``synced_at`` to ``likes`` and ``play_history`` — the server-side ingestion
+time used as the delta-sync ordering key (distinct from the event time
+``created_at``/``played_at``, which a sync push preserves from the client even
+when the event happened offline earlier). Existing rows are backfilled from
+their event time so a first sync after upgrade behaves sensibly.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260728_sync_synced_at"
+down_revision: str | None = "20260728_user_settings"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "likes",
+        sa.Column(
+            "synced_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.execute("UPDATE likes SET synced_at = created_at")
+    op.create_index(op.f("ix_likes_synced_at"), "likes", ["synced_at"])
+
+    op.add_column(
+        "play_history",
+        sa.Column(
+            "synced_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.execute("UPDATE play_history SET synced_at = played_at")
+    op.create_index(op.f("ix_play_history_synced_at"), "play_history", ["synced_at"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_play_history_synced_at"), table_name="play_history")
+    op.drop_column("play_history", "synced_at")
+    op.drop_index(op.f("ix_likes_synced_at"), table_name="likes")
+    op.drop_column("likes", "synced_at")

app/api/deps.py

@@ -6,12 +6,14 @@ bound to the request-scoped DB session; stateless adapters (hasher, token
 service) are process-cached.
 """
 
+import datetime as dt
 from collections.abc import AsyncIterator
 from functools import lru_cache
 from typing import Annotated
 
 from fastapi import Depends, Query
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.application.auth_service import AuthService
@@ -20,8 +22,10 @@ from app.application.metadata_service import MetadataEnrichmentService
 from app.application.remote_library_service import RemoteLibraryService
 from app.application.streaming_service import StreamingService
 from app.application.subsonic_auth_service import SubsonicAuthService
+from app.application.sync_service import SyncService
 from app.application.upload_service import UploadService
 from app.application.user_service import UserService
+from app.application.user_settings_service import UserSettingsService
 from app.core.config import get_settings
 from app.core.security import Argon2PasswordHasher, JwtTokenService, SubsonicPasswordCipher
 from app.domain.entities import User
@@ -38,6 +42,7 @@ from app.infrastructure.db.repositories import (
     SqlAlchemyRefreshTokenRepository,
     SqlAlchemyTrackRepository,
     SqlAlchemyUserRepository,
+    SqlAlchemyUserSettingsRepository,
 )
 from app.infrastructure.metadata.acoustid import AcoustIdHttpClient
 from app.infrastructure.metadata.fingerprint import FpcalcFingerprinter
@@ -112,9 +117,17 @@ def get_subsonic_auth_service(session: SessionDep) -> SubsonicAuthService:
     )
 
 
+def get_user_settings_service(session: SessionDep) -> UserSettingsService:
+    return UserSettingsService(
+        settings=SqlAlchemyUserSettingsRepository(session),
+        cipher=get_subsonic_cipher(),
+    )
+
+
 AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
 UserServiceDep = Annotated[UserService, Depends(get_user_service)]
 SubsonicAuthServiceDep = Annotated[SubsonicAuthService, Depends(get_subsonic_auth_service)]
+UserSettingsServiceDep = Annotated[UserSettingsService, Depends(get_user_settings_service)]
 
 
 # -- file storage (process-cached) ---------------------------------------------
@@ -189,6 +202,24 @@ DownloadServiceDep = Annotated[DownloadService, Depends(get_download_service)]
 RemoteLibraryServiceDep = Annotated[RemoteLibraryService, Depends(get_remote_library_service)]
 
 
+async def _db_now(session: AsyncSession) -> dt.datetime:
+    """The database clock — the sync cursor watermark (avoids app/DB skew)."""
+    return (await session.execute(select(func.now()))).scalar_one()
+
+
+def get_sync_service(session: SessionDep) -> SyncService:
+    return SyncService(
+        likes=SqlAlchemyLikeRepository(session),
+        history=SqlAlchemyHistoryRepository(session),
+        playlists=SqlAlchemyPlaylistRepository(session),
+        tracks=SqlAlchemyTrackRepository(session),
+        now=lambda: _db_now(session),
+    )
+
+
+SyncServiceDep = Annotated[SyncService, Depends(get_sync_service)]
+
+
 # -- library repository deps ---------------------------------------------------
 def get_track_repository(session: SessionDep) -> SqlAlchemyTrackRepository:
     return SqlAlchemyTrackRepository(session)

app/api/errors.py

@@ -18,6 +18,7 @@ from app.domain.errors import (
     DependencyUnavailableError,
     DomainError,
     NotFoundError,
+    NotSupportedError,
     PermissionDeniedError,
     RangeNotSatisfiableError,
     StorageError,
@@ -33,6 +34,7 @@ _STATUS_BY_ERROR: dict[type[DomainError], int] = {
     ValidationError: status.HTTP_422_UNPROCESSABLE_CONTENT,
     AuthenticationError: status.HTTP_401_UNAUTHORIZED,
     PermissionDeniedError: status.HTTP_403_FORBIDDEN,
+    NotSupportedError: status.HTTP_501_NOT_IMPLEMENTED,
     DependencyUnavailableError: status.HTTP_503_SERVICE_UNAVAILABLE,
     StorageError: status.HTTP_500_INTERNAL_SERVER_ERROR,
 }

app/api/schemas/admin.py

@@ -0,0 +1,39 @@
+"""Admin (instance-management) response schemas."""
+
+from pydantic import BaseModel
+
+from app.api.health import CheckStatus
+
+
+class ServicesStatusOut(BaseModel):
+    """Backing-dependency health for the admin dashboard (mirrors readiness)."""
+
+    database: CheckStatus
+    redis: CheckStatus
+    ml: CheckStatus
+
+
+class ReindexJob(BaseModel):
+    source: str
+    job_id: str
+
+
+class ReindexResponse(BaseModel):
+    """The scan jobs enqueued by a re-index, one per indexable source."""
+
+    jobs: list[ReindexJob]
+
+
+class AdminSettingsOut(BaseModel):
+    """Effective, non-secret instance configuration. Secrets and connection
+    strings are never exposed — only whether an optional integration is set up."""
+
+    environment: str
+    allow_registration: bool
+    storage_backend: str
+    media_path: str
+    youtube_enabled: bool
+    coverart_enabled: bool
+    ml_configured: bool
+    acoustid_configured: bool
+    local_import_configured: bool

app/api/schemas/settings.py

@@ -0,0 +1,49 @@
+"""User-settings request/response schemas.
+
+Enums are enforced at the API boundary (Pydantic ``Literal`` → 422 on bad
+input), so the service can trust the values it receives.
+"""
+
+from typing import Literal
+
+from pydantic import BaseModel, model_validator
+
+Theme = Literal["system", "light", "dark"]
+# Playback quality preference. ``original`` = no transcode; the lower tiers are
+# consumed by the (upcoming) transcoding pipeline.
+StreamQuality = Literal["original", "high", "medium", "low"]
+ScrobbleProvider = Literal["lastfm", "listenbrainz"]
+
+
+class SettingsOut(BaseModel):
+    theme: Theme
+    stream_quality: StreamQuality
+
+
+class SettingsUpdate(BaseModel):
+    """Partial update — omitted fields keep their current value."""
+
+    theme: Theme | None = None
+    stream_quality: StreamQuality | None = None
+
+
+class ScrobblingOut(BaseModel):
+    enabled: bool
+    provider: ScrobbleProvider | None
+    username: str | None
+    # Whether a session key is stored. The key itself is never returned.
+    configured: bool
+
+
+class ScrobblingUpdate(BaseModel):
+    enabled: bool = False
+    provider: ScrobbleProvider | None = None
+    username: str | None = None
+    # Write-only scrobbler session key / user token. Omit to keep the stored one.
+    session_key: str | None = None
+
+    @model_validator(mode="after")
+    def _provider_required_when_enabled(self) -> ScrobblingUpdate:
+        if self.enabled and self.provider is None:
+            raise ValueError("provider is required when scrobbling is enabled")
+        return self

app/api/schemas/storage.py

@@ -4,6 +4,8 @@ import datetime as dt
 
 from pydantic import BaseModel
 
+from app.api.schemas.track import TrackOut
+
 
 class DiskUsageOut(BaseModel):
     total: int
@@ -43,3 +45,17 @@ class StorageStatsOut(BaseModel):
 
     # backing volume (``None`` for object-store backends)
     disk: DiskUsageOut | None
+
+
+class DuplicateGroupOut(BaseModel):
+    """Tracks sharing one acoustic fingerprint — candidates for de-duplication."""
+
+    fingerprint: str
+    tracks: list[TrackOut]
+
+
+class CleanupEnqueuedOut(BaseModel):
+    """Acknowledgement that a cleanup job was queued (it runs in the worker)."""
+
+    status: str
+    job_id: str

app/api/schemas/sync.py

@@ -0,0 +1,82 @@
+"""Offline-first sync schemas (delta pull + idempotent push).
+
+The client keeps an opaque ``cursor`` (a server-clock timestamp). It pulls
+everything changed in the half-open window ``(since, cursor]`` and pushes the
+append-only events it accumulated offline. Events carry a client-generated
+``id`` so a replay is idempotent.
+"""
+
+import datetime as dt
+import uuid
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from app.api.schemas.track import TrackOut
+
+LikeValue = Literal["like", "dislike", "neutral"]
+
+
+# -- pull (server -> client) --------------------------------------------------
+class LikeEventOut(BaseModel):
+    id: uuid.UUID
+    track_id: uuid.UUID
+    value: str
+    created_at: dt.datetime
+
+
+class PlayEventOut(BaseModel):
+    id: uuid.UUID
+    track_id: uuid.UUID
+    played_at: dt.datetime
+    play_duration_seconds: int | None
+    completed: bool
+
+
+class PlaylistSyncOut(BaseModel):
+    id: uuid.UUID
+    name: str
+    description: str | None
+    version: int
+    updated_at: dt.datetime
+    track_ids: list[uuid.UUID]
+
+
+class SyncChangesOut(BaseModel):
+    """Everything that changed for the caller since their last cursor. Feed
+    ``cursor`` back as ``?since=`` on the next pull."""
+
+    cursor: dt.datetime
+    likes: list[LikeEventOut]
+    plays: list[PlayEventOut]
+    playlists: list[PlaylistSyncOut]
+    tracks: list[TrackOut]
+
+
+# -- push (client -> server) --------------------------------------------------
+class LikeEventIn(BaseModel):
+    id: uuid.UUID
+    track_id: uuid.UUID
+    value: LikeValue
+    created_at: dt.datetime
+
+
+class PlayEventIn(BaseModel):
+    id: uuid.UUID
+    track_id: uuid.UUID
+    played_at: dt.datetime
+    play_duration_seconds: int | None = None
+    completed: bool = False
+
+
+class SyncPushIn(BaseModel):
+    likes: list[LikeEventIn] = Field(default_factory=list)
+    plays: list[PlayEventIn] = Field(default_factory=list)
+
+
+class SyncPushOut(BaseModel):
+    """How many events were newly stored (a replay reports 0) + a fresh cursor."""
+
+    cursor: dt.datetime
+    accepted_likes: int
+    accepted_plays: int

app/api/v1/admin.py

@@ -5,11 +5,18 @@ sign-up (plan §6.4).
 """
 
 import uuid
-from typing import Any
 
 from fastapi import APIRouter, Query, status
 
-from app.api.deps import SubsonicAuthServiceDep, SuperUser, UserServiceDep
+from app.api.deps import SourceRegistryDep, SubsonicAuthServiceDep, SuperUser, UserServiceDep
+from app.api.health import _check_db, _check_ml, _check_redis
+from app.api.schemas.admin import (
+    AdminSettingsOut,
+    ReindexJob,
+    ReindexResponse,
+    ServicesStatusOut,
+)
+from app.api.schemas.source import SourceInfoOut
 from app.api.schemas.subsonic import SubsonicPasswordResponse
 from app.api.schemas.user import (
     CreateUserRequest,
@@ -17,6 +24,9 @@ from app.api.schemas.user import (
     UpdateUserRequest,
     UserResponse,
 )
+from app.core.config import get_settings
+from app.domain.errors import DependencyUnavailableError, NotSupportedError
+from app.workers.queue import enqueue
 
 router = APIRouter(prefix="/admin", tags=["admin"])
 
@@ -91,24 +101,74 @@ async def rotate_user_subsonic_password(
 
 
 @router.get("/services")
-async def list_services(_admin: SuperUser) -> Any: ...
+async def list_services(_admin: SuperUser) -> ServicesStatusOut:
+    """Backing-dependency health for the admin dashboard — same probes as the
+    readiness endpoint (DB + Redis required, ML optional)."""
+    database = await _check_db()
+    redis = await _check_redis()
+    ml = await _check_ml()
+    return ServicesStatusOut(database=database, redis=redis, ml=ml)
 
 
 @router.get("/sources")
-async def list_admin_sources(_admin: SuperUser) -> Any: ...
-
-
-@router.patch("/sources/{source}")
-async def update_admin_source(source: str, _admin: SuperUser) -> Any: ...
+async def list_admin_sources(
+    _admin: SuperUser, registry: SourceRegistryDep
+) -> list[SourceInfoOut]:
+    """Configured sources and their live availability (same view as
+    ``/sources``, admin-scoped)."""
+    return [SourceInfoOut.from_entity(info) for info in registry.infos()]
 
 
 @router.post("/reindex")
-async def trigger_reindex(_admin: SuperUser) -> Any: ...
+async def trigger_reindex(admin: SuperUser, registry: SourceRegistryDep) -> ReindexResponse:
+    """Enqueue a full re-scan of every indexable source. The walk + file copies
+    run in the worker (never the request cycle); re-scans are idempotent."""
+    indexables = registry.indexables()
+    if not indexables:
+        raise DependencyUnavailableError("No indexable source is configured.")
+    jobs: list[ReindexJob] = []
+    for backend in indexables:
+        job_id = await enqueue("scan_local_folder", source=backend.name, added_by=str(admin.id))
+        jobs.append(ReindexJob(source=backend.name, job_id=job_id))
+    return ReindexResponse(jobs=jobs)
 
 
 @router.get("/settings")
-async def get_admin_settings(_admin: SuperUser) -> Any: ...
+async def get_admin_settings(_admin: SuperUser) -> AdminSettingsOut:
+    """Effective, non-secret instance configuration. Reflects the environment the
+    process booted with; secrets/connection strings are never returned — only
+    whether each optional integration is configured."""
+    settings = get_settings()
+    return AdminSettingsOut(
+        environment=settings.environment,
+        allow_registration=settings.allow_registration,
+        storage_backend=settings.storage_backend,
+        media_path=str(settings.media_path),
+        youtube_enabled=settings.youtube_enabled,
+        coverart_enabled=settings.coverart_enabled,
+        ml_configured=settings.ml_service_url is not None,
+        acoustid_configured=settings.acoustid_api_key is not None,
+        local_import_configured=settings.local_media_import_path is not None,
+    )
+
+
+# -- runtime config mutation (intentionally unsupported) ----------------------
+# The instance is env-configured (CLAUDE.md: nothing hardcoded, all from env) and
+# get_settings() is a cached singleton, so config is not mutable at runtime.
+# These endpoints answer 501 with a clear reason rather than silently no-op'ing;
+# a persistent override layer that shadows env would be a deliberate future
+# departure. Read the effective config via GET /admin/settings.
+@router.patch("/sources/{source}")
+async def update_admin_source(source: str, _admin: SuperUser) -> None:
+    raise NotSupportedError(
+        "Sources are configured via environment variables (e.g. YOUTUBE_ENABLED, "
+        "LOCAL_MEDIA_IMPORT_PATH); runtime changes are not supported."
+    )
 
 
 @router.patch("/settings")
-async def update_admin_settings(_admin: SuperUser) -> Any: ...
+async def update_admin_settings(_admin: SuperUser) -> None:
+    raise NotSupportedError(
+        "Instance settings are managed via environment configuration; "
+        "runtime changes are not supported."
+    )

app/api/v1/playlists.py

@@ -1,15 +1,18 @@
 """Playlist endpoints."""
 
 import uuid
-from typing import Any
 
 from fastapi import APIRouter, Query, Response
+from fastapi.responses import StreamingResponse
 
+from app.api.covers import stream_cover
 from app.api.deps import (
     AlbumRepoDep,
     ArtistRepoDep,
     CurrentUser,
+    FileStorageDep,
     PlaylistRepoDep,
+    StreamUser,
     TrackRepoDep,
 )
 from app.api.schemas.pagination import PagedResponse
@@ -217,4 +220,14 @@ async def reorder_playlist_tracks(
 
 
 @router.get("/{playlist_id}/cover")
-async def get_playlist_cover(playlist_id: uuid.UUID, _: CurrentUser) -> Any: ...
+async def get_playlist_cover(
+    playlist_id: uuid.UUID,
+    playlist_repo: PlaylistRepoDep,
+    storage: FileStorageDep,
+    _: StreamUser,
+) -> StreamingResponse:
+    # ``<img>`` can't send a bearer header → StreamUser accepts ``?token=``.
+    cover_path = await playlist_repo.get_cover_path(playlist_id)
+    if not cover_path:
+        raise NotFoundError("Cover not found.")
+    return await stream_cover(storage, cover_path)

app/api/v1/storage.py

@@ -1,22 +1,28 @@
 """Storage analysis and cleanup endpoints."""
 
-from typing import Any
-
-from fastapi import APIRouter
+from fastapi import APIRouter, Query
 
 from app.api.deps import (
     AlbumRepoDep,
     ArtistRepoDep,
     CurrentUser,
     FileStorageDep,
+    SuperUser,
     TrackRepoDep,
 )
+from app.api.schemas.pagination import PagedResponse
 from app.api.schemas.storage import (
+    CleanupEnqueuedOut,
     DiskUsageOut,
+    DuplicateGroupOut,
     FormatBreakdownOut,
     GenreCountOut,
     StorageStatsOut,
 )
+from app.api.schemas.track import TrackOut
+from app.api.v1.tracks import _build_track_out
+from app.domain.entities.track import Track
+from app.workers.queue import enqueue
 
 router = APIRouter(prefix="/storage", tags=["storage"])
 
@@ -24,6 +30,18 @@ router = APIRouter(prefix="/storage", tags=["storage"])
 _TOP_GENRES = 8
 
 
+async def _tracks_to_out(
+    tracks: list[Track], artist_repo: ArtistRepoDep, album_repo: AlbumRepoDep
+) -> list[TrackOut]:
+    """Hydrate a batch of tracks into ``TrackOut`` (artist/album names + cover
+    flag), resolving each referenced artist/album in a single query."""
+    artist_ids = list({t.artist_id for t in tracks})
+    album_ids = list({t.album_id for t in tracks if t.album_id is not None})
+    artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
+    albums = {a.id: a for a in await album_repo.get_many(album_ids)}
+    return await _build_track_out(tracks, artists, albums)
+
+
 @router.get("")
 async def get_storage_stats(
     track_repo: TrackRepoDep,
@@ -70,16 +88,63 @@ async def get_storage_stats(
 
 
 @router.get("/duplicates")
-async def get_duplicates() -> Any: ...
+async def get_duplicates(
+    track_repo: TrackRepoDep,
+    artist_repo: ArtistRepoDep,
+    album_repo: AlbumRepoDep,
+    _: CurrentUser,
+) -> list[DuplicateGroupOut]:
+    """Tracks sharing an acoustic fingerprint, grouped — the library's real
+    duplicates (``(source, source_id)`` is already unique). Cheap DB GROUP BY."""
+    groups = await track_repo.find_duplicate_groups()
+    all_tracks = [track for _, tracks in groups for track in tracks]
+    out = await _tracks_to_out(all_tracks, artist_repo, album_repo)
+    by_id = {item.id: item for item in out}
+    return [
+        DuplicateGroupOut(fingerprint=fingerprint, tracks=[by_id[t.id] for t in tracks])
+        for fingerprint, tracks in groups
+    ]
 
 
 @router.get("/broken")
-async def get_broken_files() -> Any: ...
+async def get_broken_files(
+    track_repo: TrackRepoDep,
+    artist_repo: ArtistRepoDep,
+    album_repo: AlbumRepoDep,
+    _: CurrentUser,
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+) -> PagedResponse[TrackOut]:
+    """Tracks whose last enrichment run failed (``metadata_status=failed``) —
+    each carries its ``metadata_error``. A file gone missing on disk is instead
+    reconciled by ``POST /storage/cleanup`` (that needs a filesystem scan)."""
+    tracks = await track_repo.list_by_metadata_status("failed", limit=limit, offset=offset)
+    total = await track_repo.count_by_metadata_status("failed")
+    items = await _tracks_to_out(tracks, artist_repo, album_repo)
+    return PagedResponse(items=items, total=total, limit=limit, offset=offset)
 
 
 @router.get("/missing-metadata")
-async def get_missing_metadata() -> Any: ...
+async def get_missing_metadata(
+    track_repo: TrackRepoDep,
+    artist_repo: ArtistRepoDep,
+    album_repo: AlbumRepoDep,
+    _: CurrentUser,
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+) -> PagedResponse[TrackOut]:
+    """Tracks still awaiting enrichment (``metadata_status=pending``) — imported
+    but never identified."""
+    tracks = await track_repo.list_by_metadata_status("pending", limit=limit, offset=offset)
+    total = await track_repo.count_by_metadata_status("pending")
+    items = await _tracks_to_out(tracks, artist_repo, album_repo)
+    return PagedResponse(items=items, total=total, limit=limit, offset=offset)
 
 
-@router.post("/cleanup")
-async def run_cleanup() -> Any: ...
+@router.post("/cleanup", status_code=202)
+async def run_cleanup(_: SuperUser) -> CleanupEnqueuedOut:
+    """Admin: enqueue the storage reconciliation job. It scans the catalogue and
+    removes rows whose backing file has vanished (dangling references). Runs in
+    the worker — the filesystem scan must not block the request cycle."""
+    job_id = await enqueue("cleanup_storage")
+    return CleanupEnqueuedOut(status="enqueued", job_id=job_id)

app/api/v1/sync.py

@@ -1,15 +1,103 @@
-"""Client sync endpoints (offline-first event log)."""
+"""Client sync endpoints (offline-first event log).
 
-from typing import Any
+``GET /sync/changes`` pulls everything the caller changed since their cursor;
+``POST /sync/push`` uploads the like/play events a client accumulated offline
+(idempotent — replays are no-ops). See :mod:`app.application.sync_service`.
+"""
+
+import datetime as dt
 
 from fastapi import APIRouter
 
+from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, SyncServiceDep
+from app.api.schemas.sync import (
+    LikeEventOut,
+    PlayEventOut,
+    PlaylistSyncOut,
+    SyncChangesOut,
+    SyncPushIn,
+    SyncPushOut,
+)
+from app.api.v1.tracks import _build_track_out
+from app.application.sync_service import LikeEvent, PlayEvent
+
 router = APIRouter(prefix="/sync", tags=["sync"])
 
 
 @router.get("/changes")
-async def get_changes() -> Any: ...
+async def get_changes(
+    service: SyncServiceDep,
+    artist_repo: ArtistRepoDep,
+    album_repo: AlbumRepoDep,
+    user: CurrentUser,
+    since: dt.datetime | None = None,
+) -> SyncChangesOut:
+    """Delta since ``since`` (omit for a full snapshot). Persist ``cursor`` from
+    the response and pass it back as ``?since=`` next time."""
+    changes = await service.get_changes(user.id, since=since)
+
+    artist_ids = list({t.artist_id for t in changes.tracks})
+    album_ids = list({t.album_id for t in changes.tracks if t.album_id is not None})
+    artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
+    albums = {a.id: a for a in await album_repo.get_many(album_ids)}
+    tracks_out = await _build_track_out(changes.tracks, artists, albums)
+
+    return SyncChangesOut(
+        cursor=changes.cursor,
+        likes=[
+            LikeEventOut(
+                id=lk.id, track_id=lk.track_id, value=lk.value, created_at=lk.created_at
+            )
+            for lk in changes.likes
+        ],
+        plays=[
+            PlayEventOut(
+                id=p.id,
+                track_id=p.track_id,
+                played_at=p.played_at,
+                play_duration_seconds=p.play_duration_seconds,
+                completed=p.completed,
+            )
+            for p in changes.plays
+        ],
+        playlists=[
+            PlaylistSyncOut(
+                id=d.playlist.id,
+                name=d.playlist.name,
+                description=d.playlist.description,
+                version=d.playlist.version,
+                updated_at=d.playlist.updated_at,
+                track_ids=d.track_ids,
+            )
+            for d in changes.playlists
+        ],
+        tracks=tracks_out,
+    )
 
 
 @router.post("/push")
-async def push_changes() -> Any: ...
+async def push_changes(
+    body: SyncPushIn, service: SyncServiceDep, user: CurrentUser
+) -> SyncPushOut:
+    result = await service.push(
+        user.id,
+        likes=[
+            LikeEvent(id=e.id, track_id=e.track_id, value=e.value, created_at=e.created_at)
+            for e in body.likes
+        ],
+        plays=[
+            PlayEvent(
+                id=e.id,
+                track_id=e.track_id,
+                played_at=e.played_at,
+                play_duration_seconds=e.play_duration_seconds,
+                completed=e.completed,
+            )
+            for e in body.plays
+        ],
+    )
+    return SyncPushOut(
+        cursor=result.cursor,
+        accepted_likes=result.accepted_likes,
+        accepted_plays=result.accepted_plays,
+    )

app/api/v1/user_settings.py

@@ -1,23 +1,67 @@
-"""User settings endpoints, including scrobbling configuration."""
+"""User settings endpoints, including scrobbling configuration.
 
-from typing import Any
+Settings are per-caller and created lazily, so a first read returns defaults.
+The scrobbler session key is write-only — accepted on ``PUT`` but never returned.
+"""
 
 from fastapi import APIRouter
 
+from app.api.deps import CurrentUser, UserSettingsServiceDep
+from app.api.schemas.settings import (
+    ScrobblingOut,
+    ScrobblingUpdate,
+    SettingsOut,
+    SettingsUpdate,
+)
+from app.domain.entities.settings import UserSettings
+
 router = APIRouter(prefix="/settings", tags=["settings"])
 
 
+def _to_settings_out(settings: UserSettings) -> SettingsOut:
+    return SettingsOut(theme=settings.theme, stream_quality=settings.stream_quality)
+
+
+def _to_scrobbling_out(settings: UserSettings) -> ScrobblingOut:
+    return ScrobblingOut(
+        enabled=settings.scrobble_enabled,
+        provider=settings.scrobble_provider,
+        username=settings.scrobble_username,
+        configured=settings.scrobble_session_key_enc is not None,
+    )
+
+
 @router.get("")
-async def get_settings() -> Any: ...
+async def get_settings(user: CurrentUser, service: UserSettingsServiceDep) -> SettingsOut:
+    return _to_settings_out(await service.get(user.id))
 
 
 @router.patch("")
-async def update_settings() -> Any: ...
+async def update_settings(
+    body: SettingsUpdate, user: CurrentUser, service: UserSettingsServiceDep
+) -> SettingsOut:
+    settings = await service.update_general(
+        user.id, theme=body.theme, stream_quality=body.stream_quality
+    )
+    return _to_settings_out(settings)
 
 
 @router.get("/scrobbling")
-async def get_scrobbling_settings() -> Any: ...
+async def get_scrobbling_settings(
+    user: CurrentUser, service: UserSettingsServiceDep
+) -> ScrobblingOut:
+    return _to_scrobbling_out(await service.get(user.id))
 
 
 @router.put("/scrobbling")
-async def set_scrobbling_settings() -> Any: ...
+async def set_scrobbling_settings(
+    body: ScrobblingUpdate, user: CurrentUser, service: UserSettingsServiceDep
+) -> ScrobblingOut:
+    settings = await service.set_scrobbling(
+        user.id,
+        enabled=body.enabled,
+        provider=body.provider,
+        username=body.username,
+        session_key=body.session_key,
+    )
+    return _to_scrobbling_out(settings)

app/application/sync_service.py

@@ -0,0 +1,140 @@
+"""Offline-first sync use cases: delta pull + idempotent push.
+
+The cursor is a server-clock timestamp obtained from the DB (``now``), so it is
+immune to app/DB clock skew. A pull returns everything a user changed in the
+half-open window ``(since, cursor]``; a push appends the event-log entries a
+client accumulated offline. Events carry a client-generated id, so a replay is a
+no-op (append is ``ON CONFLICT DO NOTHING``). Events for tracks this server does
+not have are skipped rather than rejected (graceful degradation).
+
+Known limitation (v1): the delta is unpaginated and, being wall-clock based, a
+write committing right on the cursor boundary under concurrency can slip a cycle
+— a periodic full resync (``since=None``) heals it.
+"""
+
+import datetime as dt
+import uuid
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+
+from app.domain.entities.history import PlayHistoryEntry
+from app.domain.entities.like import Like
+from app.domain.entities.playlist import Playlist
+from app.domain.entities.track import Track
+from app.domain.ports import (
+    HistoryRepository,
+    LikeRepository,
+    PlaylistRepository,
+    TrackRepository,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class LikeEvent:
+    id: uuid.UUID
+    track_id: uuid.UUID
+    value: str
+    created_at: dt.datetime
+
+
+@dataclass(frozen=True, slots=True)
+class PlayEvent:
+    id: uuid.UUID
+    track_id: uuid.UUID
+    played_at: dt.datetime
+    play_duration_seconds: int | None
+    completed: bool
+
+
+@dataclass(frozen=True, slots=True)
+class PlaylistDelta:
+    playlist: Playlist
+    track_ids: list[uuid.UUID]
+
+
+@dataclass(frozen=True, slots=True)
+class SyncChanges:
+    cursor: dt.datetime
+    likes: list[Like]
+    plays: list[PlayHistoryEntry]
+    playlists: list[PlaylistDelta]
+    tracks: list[Track]
+
+
+@dataclass(frozen=True, slots=True)
+class SyncPushResult:
+    cursor: dt.datetime
+    accepted_likes: int
+    accepted_plays: int
+
+
+class SyncService:
+    def __init__(
+        self,
+        *,
+        likes: LikeRepository,
+        history: HistoryRepository,
+        playlists: PlaylistRepository,
+        tracks: TrackRepository,
+        now: Callable[[], Awaitable[dt.datetime]],
+    ) -> None:
+        self._likes = likes
+        self._history = history
+        self._playlists = playlists
+        self._tracks = tracks
+        self._now = now
+
+    async def get_changes(self, user_id: uuid.UUID, *, since: dt.datetime | None) -> SyncChanges:
+        until = await self._now()
+        likes = await self._likes.list_since(user_id, since=since, until=until)
+        plays = await self._history.list_since(user_id, since=since, until=until)
+        changed = await self._playlists.list_changed_since(
+            owner_id=user_id, since=since, until=until
+        )
+        playlists = [
+            PlaylistDelta(playlist=p, track_ids=await self._playlists.track_ids(p.id))
+            for p in changed
+        ]
+        tracks = await self._tracks.list_changed_since(since=since, until=until)
+        return SyncChanges(
+            cursor=until, likes=likes, plays=plays, playlists=playlists, tracks=tracks
+        )
+
+    async def push(
+        self,
+        user_id: uuid.UUID,
+        *,
+        likes: list[LikeEvent],
+        plays: list[PlayEvent],
+    ) -> SyncPushResult:
+        accepted_likes = 0
+        for event in likes:
+            if await self._tracks.get_by_id(event.track_id) is None:
+                continue  # skip events for tracks this server doesn't have
+            if await self._likes.add_event(
+                id=event.id,
+                user_id=user_id,
+                track_id=event.track_id,
+                value=event.value,
+                created_at=event.created_at,
+            ):
+                accepted_likes += 1
+
+        accepted_plays = 0
+        for play in plays:
+            if await self._tracks.get_by_id(play.track_id) is None:
+                continue
+            if await self._history.add_event(
+                id=play.id,
+                user_id=user_id,
+                track_id=play.track_id,
+                played_at=play.played_at,
+                play_duration_seconds=play.play_duration_seconds,
+                completed=play.completed,
+            ):
+                accepted_plays += 1
+
+        cursor = await self._now()
+        return SyncPushResult(
+            cursor=cursor, accepted_likes=accepted_likes, accepted_plays=accepted_plays
+        )

app/application/user_settings_service.py

@@ -0,0 +1,62 @@
+"""User-settings use cases: general preferences + scrobbling configuration.
+
+Settings rows are created lazily — a user who never saved anything reads clean
+defaults. Partial updates are merged against current values here, so the
+repository always persists the complete desired state. The scrobbler session
+key is encrypted before it touches the DB (never stored or returned in plain).
+"""
+
+import uuid
+from dataclasses import replace
+
+from app.domain.entities.settings import UserSettings
+from app.domain.ports import SubsonicCipher, UserSettingsRepository
+
+
+class UserSettingsService:
+    def __init__(self, *, settings: UserSettingsRepository, cipher: SubsonicCipher) -> None:
+        self._settings = settings
+        # Same Fernet cipher used for the Subsonic app-password — reused here to
+        # encrypt the scrobbler session key at rest (symmetric, recoverable).
+        self._cipher = cipher
+
+    async def get(self, user_id: uuid.UUID) -> UserSettings:
+        return await self._settings.get(user_id) or UserSettings.defaults(user_id)
+
+    async def update_general(
+        self, user_id: uuid.UUID, *, theme: str | None, stream_quality: str | None
+    ) -> UserSettings:
+        current = await self.get(user_id)
+        merged = replace(
+            current,
+            theme=theme if theme is not None else current.theme,
+            stream_quality=stream_quality if stream_quality is not None else current.stream_quality,
+        )
+        return await self._settings.upsert(merged)
+
+    async def set_scrobbling(
+        self,
+        user_id: uuid.UUID,
+        *,
+        enabled: bool,
+        provider: str | None,
+        username: str | None,
+        session_key: str | None,
+    ) -> UserSettings:
+        """Replace the scrobbling config. ``session_key`` is write-only: a new
+        value is encrypted and stored; omitting it keeps the existing key (the
+        client can't read it back to re-send it)."""
+        current = await self.get(user_id)
+        session_key_enc: str | None
+        if session_key is not None:
+            session_key_enc = self._cipher.encrypt(session_key)
+        else:
+            session_key_enc = current.scrobble_session_key_enc
+        merged = replace(
+            current,
+            scrobble_enabled=enabled,
+            scrobble_provider=provider,
+            scrobble_username=username,
+            scrobble_session_key_enc=session_key_enc,
+        )
+        return await self._settings.upsert(merged)

app/core/config.py

@@ -61,6 +61,17 @@ class Settings(BaseSettings):
     # admin-only (POST /admin/users). Registered users are never superusers.
     allow_registration: bool = True
 
+    # -- CORS -------------------------------------------------------------
+    # Origins allowed to call the API from a browser. The web UI is multi-
+    # instance — it connects to whatever origin the operator types on the
+    # connect screen — so a page served from origin A may call this backend at
+    # origin B (e.g. the direct :8000 port, a LAN IP, or 127.0.0.1 vs localhost).
+    # Auth rides in the ``Authorization`` bearer header (not cookies), so the
+    # wildcard default is safe here — it is paired with ``allow_credentials=False``.
+    # Set explicit origins in hardened deployments. Accepts a comma-separated
+    # string in ``.env`` (``CORS_ALLOW_ORIGINS=https://a,https://b``) or ``*``.
+    cors_allow_origins: list[str] = Field(default_factory=lambda: ["*"])
+
     # -- subsonic ---------------------------------------------------------
     # Symmetric key (any string) used to encrypt each user's recoverable
     # Subsonic app-password at rest. A Fernet key is derived from it; rotating
@@ -127,6 +138,15 @@ class Settings(BaseSettings):
             raise ValueError("database_url must use the asyncpg driver: postgresql+asyncpg://")
         return v
 
+    @field_validator("cors_allow_origins", mode="before")
+    @classmethod
+    def _split_cors_origins(cls, v: object) -> object:
+        # Allow a plain comma-separated string in .env (pydantic would otherwise
+        # try to JSON-decode a list field): "a, b" -> ["a", "b"]; "*" -> ["*"].
+        if isinstance(v, str):
+            return [origin.strip() for origin in v.split(",") if origin.strip()]
+        return v
+
     @property
     def is_prod(self) -> bool:
         return self.environment == "prod"

app/domain/entities/settings.py

@@ -0,0 +1,34 @@
+"""User settings domain entity (general preferences + scrobbling config)."""
+
+import uuid
+from dataclasses import dataclass
+
+# Defaults for a user who has never saved settings — the row is created lazily,
+# so reads return these before the first write.
+DEFAULT_THEME = "system"
+DEFAULT_STREAM_QUALITY = "original"
+
+
+@dataclass(frozen=True, slots=True)
+class UserSettings:
+    user_id: uuid.UUID
+    theme: str
+    stream_quality: str
+    scrobble_enabled: bool
+    scrobble_provider: str | None
+    scrobble_username: str | None
+    # Scrobbler session key / user token, encrypted at rest (never leaves the
+    # server in plaintext). ``None`` until the user configures scrobbling.
+    scrobble_session_key_enc: str | None
+
+    @classmethod
+    def defaults(cls, user_id: uuid.UUID) -> UserSettings:
+        return cls(
+            user_id=user_id,
+            theme=DEFAULT_THEME,
+            stream_quality=DEFAULT_STREAM_QUALITY,
+            scrobble_enabled=False,
+            scrobble_provider=None,
+            scrobble_username=None,
+            scrobble_session_key_enc=None,
+        )

app/domain/errors.py

@@ -54,6 +54,13 @@ class PermissionDeniedError(DomainError):
     code = "permission_denied"
 
 
+class NotSupportedError(DomainError):
+    """Operation is intentionally unsupported (e.g. a config knob that is managed
+    via environment, not mutable at runtime)."""
+
+    code = "not_supported"
+
+
 class DependencyUnavailableError(DomainError):
     """An external dependency (source, ML, MusicBrainz) is unavailable.
 

app/domain/ports.py

@@ -29,6 +29,7 @@ from app.domain.entities import (
     SubsonicCredentials,
     User,
 )
+from app.domain.entities.settings import UserSettings
 from app.domain.entities.track import Artist, Track
 from app.domain.sources import DownloadResult, RawMetadata, SearchResult, SourceFile, SourceInfo
 from app.domain.tokens import IssuedToken, TokenClaims, TokenType
@@ -56,6 +57,11 @@ class UserRepository(Protocol):
     async def set_subsonic_password_enc(self, user_id: uuid.UUID, password_enc: str) -> None: ...
 
 
+class UserSettingsRepository(Protocol):
+    async def get(self, user_id: uuid.UUID) -> UserSettings | None: ...
+    async def upsert(self, settings: UserSettings) -> UserSettings: ...
+
+
 class SubsonicCipher(Protocol):
     """Symmetric encrypt/decrypt for the recoverable Subsonic app-password."""
 
@@ -164,6 +170,15 @@ class TrackRepository(Protocol):
     # AlbumRepository below).
     async def genres(self) -> list[tuple[str, int]]: ...
     async def library_stats(self) -> LibraryStats: ...
+    async def find_duplicate_groups(self) -> list[tuple[str, list[Track]]]: ...
+    async def list_by_metadata_status(
+        self, status: str, *, limit: int, offset: int
+    ) -> list[Track]: ...
+    async def all_storage_refs(self) -> list[tuple[uuid.UUID, str]]: ...
+    async def count_by_metadata_status(self, status: str) -> int: ...
+    async def list_changed_since(
+        self, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Track]: ...
     async def list(
         self,
         *,
@@ -287,12 +302,29 @@ class PlaylistRepository(Protocol):
     async def reorder_tracks(
         self, playlist_id: uuid.UUID, ordered_track_ids: list[uuid.UUID]
     ) -> None: ...
+    async def get_cover_path(self, playlist_id: uuid.UUID) -> str | None: ...
+    async def list_changed_since(
+        self, *, owner_id: uuid.UUID, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Playlist]: ...
+    async def track_ids(self, playlist_id: uuid.UUID) -> list[uuid.UUID]: ...
     # list must come after any method using list[...] in its signature (name shadowing)
     async def list(self, *, owner_id: uuid.UUID, limit: int, offset: int) -> list[Playlist]: ...
 
 
 class LikeRepository(Protocol):
     async def add(self, *, user_id: uuid.UUID, track_id: uuid.UUID, value: str) -> Like: ...
+    async def add_event(
+        self,
+        *,
+        id: uuid.UUID,
+        user_id: uuid.UUID,
+        track_id: uuid.UUID,
+        value: str,
+        created_at: dt.datetime,
+    ) -> bool: ...
+    async def list_since(
+        self, user_id: uuid.UUID, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Like]: ...
     async def get_latest_state(
         self, *, user_id: uuid.UUID, track_ids: list[uuid.UUID]
     ) -> list[Like]: ...
@@ -312,6 +344,19 @@ class HistoryRepository(Protocol):
         play_duration_seconds: int | None,
         completed: bool,
     ) -> PlayHistoryEntry: ...
+    async def add_event(
+        self,
+        *,
+        id: uuid.UUID,
+        user_id: uuid.UUID,
+        track_id: uuid.UUID,
+        played_at: dt.datetime,
+        play_duration_seconds: int | None,
+        completed: bool,
+    ) -> bool: ...
+    async def list_since(
+        self, user_id: uuid.UUID, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[PlayHistoryEntry]: ...
     async def list(
         self, *, user_id: uuid.UUID, limit: int, offset: int
     ) -> list[PlayHistoryEntry]: ...

app/infrastructure/db/models/__init__.py

@@ -14,6 +14,7 @@ from app.infrastructure.db.models.play_history import PlayHistoryModel
 from app.infrastructure.db.models.playlist import PlaylistModel, PlaylistTrackModel
 from app.infrastructure.db.models.track import TrackModel
 from app.infrastructure.db.models.user import RefreshTokenModel, UserModel
+from app.infrastructure.db.models.user_settings import UserSettingsModel
 
 __all__ = [
     "AlbumModel",
@@ -27,4 +28,5 @@ __all__ = [
     "RefreshTokenModel",
     "TrackModel",
     "UserModel",
+    "UserSettingsModel",
 ]

app/infrastructure/db/models/like.py

@@ -37,3 +37,12 @@ class LikeModel(UUIDPrimaryKeyMixin, Base):
         server_default=func.now(),
         nullable=False,
     )
+    # Server ingestion time — the delta-sync ordering key. Set once at insert and
+    # never changed; distinct from ``created_at`` (the real event time, which a
+    # sync push preserves from the client even when it happened offline earlier).
+    synced_at: Mapped[dt.datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+        index=True,
+    )

app/infrastructure/db/models/play_history.py

@@ -34,3 +34,11 @@ class PlayHistoryModel(UUIDPrimaryKeyMixin, Base):
     )
     play_duration_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
     completed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    # Server ingestion time — the delta-sync ordering key (see LikeModel). Distinct
+    # from ``played_at`` (the real play time), which a sync push preserves.
+    synced_at: Mapped[dt.datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+        index=True,
+    )

app/infrastructure/db/models/user_settings.py

@@ -0,0 +1,29 @@
+"""ORM model for per-user settings (general preferences + scrobbling)."""
+
+import uuid
+
+from sqlalchemy import Boolean, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.infrastructure.db.base import Base
+from app.infrastructure.db.models.mixins import TimestampMixin
+
+
+class UserSettingsModel(TimestampMixin, Base):
+    """One row per user, created lazily on first save. The primary key *is* the
+    user id (a 1:1 extension of ``users``), so there's no separate surrogate id."""
+
+    __tablename__ = "user_settings"
+
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    theme: Mapped[str] = mapped_column(String(16), default="system", nullable=False)
+    stream_quality: Mapped[str] = mapped_column(String(16), default="original", nullable=False)
+    scrobble_enabled: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    scrobble_provider: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    scrobble_username: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    # Fernet-encrypted scrobbler session key / token (see core.security). Never
+    # the plaintext — mirrors how the Subsonic app-password is stored.
+    scrobble_session_key_enc: Mapped[str | None] = mapped_column(String(512), nullable=True)

app/infrastructure/db/repositories/__init__.py

@@ -13,6 +13,9 @@ from app.infrastructure.db.repositories.refresh_token_repository import (
 )
 from app.infrastructure.db.repositories.track_repository import SqlAlchemyTrackRepository
 from app.infrastructure.db.repositories.user_repository import SqlAlchemyUserRepository
+from app.infrastructure.db.repositories.user_settings_repository import (
+    SqlAlchemyUserSettingsRepository,
+)
 
 __all__ = [
     "SqlAlchemyAlbumRepository",
@@ -24,4 +27,5 @@ __all__ = [
     "SqlAlchemyRefreshTokenRepository",
     "SqlAlchemyTrackRepository",
     "SqlAlchemyUserRepository",
+    "SqlAlchemyUserSettingsRepository",
 ]

app/infrastructure/db/repositories/history_repository.py

@@ -4,6 +4,7 @@ import datetime as dt
 import uuid
 
 from sqlalchemy import func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.domain.entities.history import PlayHistoryEntry
@@ -46,6 +47,51 @@ class SqlAlchemyHistoryRepository:
         await self._session.refresh(row)
         return _to_entity(row)
 
+    async def add_event(
+        self,
+        *,
+        id: uuid.UUID,
+        user_id: uuid.UUID,
+        track_id: uuid.UUID,
+        played_at: dt.datetime,
+        play_duration_seconds: int | None,
+        completed: bool,
+    ) -> bool:
+        """Idempotent append for sync push: insert a client-generated play event,
+        skipping it if the ``id`` already exists (a replay). Returns whether a new
+        row was stored. Defined before ``list`` (name-shadowing)."""
+        stmt = (
+            pg_insert(PlayHistoryModel)
+            .values(
+                id=id,
+                user_id=user_id,
+                track_id=track_id,
+                played_at=played_at,
+                play_duration_seconds=play_duration_seconds,
+                completed=completed,
+            )
+            .on_conflict_do_nothing(index_elements=["id"])
+            .returning(PlayHistoryModel.id)
+        )
+        inserted = (await self._session.execute(stmt)).scalar_one_or_none()
+        return inserted is not None
+
+    async def list_since(
+        self, user_id: uuid.UUID, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[PlayHistoryEntry]:
+        """Play events for a user ingested in the half-open window ``(since,
+        until]`` (by ``synced_at``, the server-side sync key — so events pushed
+        with an older ``played_at`` still surface), oldest first. ``since=None``
+        returns everything up to ``until``."""
+        stmt = select(PlayHistoryModel).where(
+            PlayHistoryModel.user_id == user_id, PlayHistoryModel.synced_at <= until
+        )
+        if since is not None:
+            stmt = stmt.where(PlayHistoryModel.synced_at > since)
+        stmt = stmt.order_by(PlayHistoryModel.synced_at)
+        rows = (await self._session.execute(stmt)).scalars().all()
+        return [_to_entity(r) for r in rows]
+
     async def list(self, *, user_id: uuid.UUID, limit: int, offset: int) -> list[PlayHistoryEntry]:
         rows = (
             (

app/infrastructure/db/repositories/like_repository.py

@@ -3,9 +3,11 @@
 Likes are an append-only event log. Current state = latest event per (user, track).
 """
 
+import datetime as dt
 import uuid
 
 from sqlalchemy import func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.domain.entities.like import Like
@@ -59,6 +61,50 @@ class SqlAlchemyLikeRepository:
         await self._session.refresh(row)
         return _to_entity(row)
 
+    async def add_event(
+        self,
+        *,
+        id: uuid.UUID,
+        user_id: uuid.UUID,
+        track_id: uuid.UUID,
+        value: str,
+        created_at: dt.datetime,
+    ) -> bool:
+        """Idempotent append for sync push: insert a client-generated like event,
+        skipping it if the ``id`` already exists (a replay). Preserves the
+        client's ``created_at`` (the event happened offline earlier). Returns
+        whether a new row was stored."""
+        stmt = (
+            pg_insert(LikeModel)
+            .values(
+                id=id,
+                user_id=user_id,
+                track_id=track_id,
+                value=value,
+                created_at=created_at,
+            )
+            .on_conflict_do_nothing(index_elements=["id"])
+            .returning(LikeModel.id)
+        )
+        inserted = (await self._session.execute(stmt)).scalar_one_or_none()
+        return inserted is not None
+
+    async def list_since(
+        self, user_id: uuid.UUID, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Like]:
+        """Like events for a user ingested in the half-open window ``(since,
+        until]`` (by ``synced_at``, the server-side sync key — so events pushed
+        with an older ``created_at`` still surface), oldest first. ``since=None``
+        returns everything up to ``until``."""
+        stmt = select(LikeModel).where(
+            LikeModel.user_id == user_id, LikeModel.synced_at <= until
+        )
+        if since is not None:
+            stmt = stmt.where(LikeModel.synced_at > since)
+        stmt = stmt.order_by(LikeModel.synced_at)
+        rows = (await self._session.execute(stmt)).scalars().all()
+        return [_to_entity(r) for r in rows]
+
     async def get_latest_state(
         self, *, user_id: uuid.UUID, track_ids: list[uuid.UUID]
     ) -> list[Like]:

app/infrastructure/db/repositories/playlist_repository.py

@@ -1,5 +1,6 @@
 """Playlist repository — adapter over ``AsyncSession``."""
 
+import datetime as dt
 import uuid
 
 from sqlalchemy import func, select
@@ -207,6 +208,41 @@ class SqlAlchemyPlaylistRepository:
         ).scalar_one_or_none()
         return float(result) if result is not None else 0.0
 
+    async def get_cover_path(self, playlist_id: uuid.UUID) -> str | None:
+        """The playlist's stored cover key, or ``None`` (missing playlist or no
+        cover). Read directly — the entity doesn't carry the storage key."""
+        return (
+            await self._session.execute(
+                select(PlaylistModel.cover_path).where(PlaylistModel.id == playlist_id)
+            )
+        ).scalar_one_or_none()
+
+    async def list_changed_since(
+        self, *, owner_id: uuid.UUID, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Playlist]:
+        """A user's playlists changed in the window ``(since, until]`` (metadata
+        or membership — every mutation bumps ``updated_at``/``version``). Defined
+        before ``list`` (name-shadowing)."""
+        stmt = select(PlaylistModel).where(
+            PlaylistModel.owner_id == owner_id, PlaylistModel.updated_at <= until
+        )
+        if since is not None:
+            stmt = stmt.where(PlaylistModel.updated_at > since)
+        stmt = stmt.order_by(PlaylistModel.updated_at)
+        rows = (await self._session.execute(stmt)).scalars().all()
+        return [_to_entity(r) for r in rows]
+
+    async def track_ids(self, playlist_id: uuid.UUID) -> list[uuid.UUID]:
+        """Ordered track ids of a playlist (by position) — for sync payloads."""
+        rows = (
+            await self._session.execute(
+                select(PlaylistTrackModel.track_id)
+                .where(PlaylistTrackModel.playlist_id == playlist_id)
+                .order_by(PlaylistTrackModel.position)
+            )
+        ).scalars().all()
+        return list(rows)
+
     # list must come after methods using list[...] in signatures (builtin name shadowing)
     async def list(self, *, owner_id: uuid.UUID, limit: int, offset: int) -> list[Playlist]:
         rows = (

app/infrastructure/db/repositories/track_repository.py

@@ -194,6 +194,91 @@ class SqlAlchemyTrackRepository:
             by_source={source: cnt for source, cnt in source_rows},
         )
 
+    async def find_duplicate_groups(self) -> list[tuple[str, list[Track]]]:
+        """Tracks that share an ``acoustid_fingerprint`` (the dedup key), grouped
+        by it — only fingerprints with more than one track. Empty when clean.
+        Defined before ``list`` for the same name-shadowing reason as ``genres``."""
+        dup_fps = (
+            select(TrackModel.acoustid_fingerprint)
+            .where(TrackModel.acoustid_fingerprint.is_not(None))
+            .group_by(TrackModel.acoustid_fingerprint)
+            .having(func.count(TrackModel.id) > 1)
+            .scalar_subquery()
+        )
+        rows = (
+            (
+                await self._session.execute(
+                    select(TrackModel)
+                    .where(TrackModel.acoustid_fingerprint.in_(dup_fps))
+                    .order_by(TrackModel.acoustid_fingerprint, TrackModel.created_at)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        groups: dict[str, list[Track]] = {}
+        for row in rows:
+            fingerprint = row.acoustid_fingerprint
+            assert fingerprint is not None  # filtered to non-null above
+            groups.setdefault(fingerprint, []).append(_to_entity(row))
+        return list(groups.items())
+
+    async def list_by_metadata_status(
+        self, status: str, *, limit: int, offset: int
+    ) -> list[Track]:
+        """Tracks in a given ``metadata_status`` (e.g. ``pending``/``failed``),
+        newest first. Defined before ``list`` (name-shadowing)."""
+        rows = (
+            (
+                await self._session.execute(
+                    select(TrackModel)
+                    .where(TrackModel.metadata_status == status)
+                    .order_by(TrackModel.created_at.desc())
+                    .limit(limit)
+                    .offset(offset)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        return [_to_entity(r) for r in rows]
+
+    async def all_storage_refs(self) -> list[tuple[uuid.UUID, str]]:
+        """``(id, storage_uri)`` for every *local* track — for the cleanup
+        worker's filesystem reconciliation. Remote placeholders have no local
+        file (``availability != local``) and are skipped. No entity hydration."""
+        rows = (
+            await self._session.execute(
+                select(TrackModel.id, TrackModel.storage_uri).where(
+                    TrackModel.availability == TrackAvailability.LOCAL.value,
+                    TrackModel.storage_uri.is_not(None),
+                )
+            )
+        ).all()
+        return [(row.id, row.storage_uri) for row in rows]
+
+    async def count_by_metadata_status(self, status: str) -> int:
+        return (
+            await self._session.execute(
+                select(func.count())
+                .select_from(TrackModel)
+                .where(TrackModel.metadata_status == status)
+            )
+        ).scalar_one()
+
+    async def list_changed_since(
+        self, *, since: dt.datetime | None, until: dt.datetime
+    ) -> list[Track]:
+        """Catalogue tracks changed in the window ``(since, until]`` (by
+        ``updated_at``), oldest first — the delta a client caches for offline use.
+        Defined before ``list`` (name-shadowing)."""
+        stmt = select(TrackModel).where(TrackModel.updated_at <= until)
+        if since is not None:
+            stmt = stmt.where(TrackModel.updated_at > since)
+        stmt = stmt.order_by(TrackModel.updated_at)
+        rows = (await self._session.execute(stmt)).scalars().all()
+        return [_to_entity(r) for r in rows]
+
     async def list(
         self,
         *,

app/infrastructure/db/repositories/user_settings_repository.py

@@ -0,0 +1,47 @@
+"""User-settings repository — adapter over ``AsyncSession``."""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.domain.entities.settings import UserSettings
+from app.infrastructure.db.models.user_settings import UserSettingsModel
+
+
+def _to_entity(row: UserSettingsModel) -> UserSettings:
+    return UserSettings(
+        user_id=row.user_id,
+        theme=row.theme,
+        stream_quality=row.stream_quality,
+        scrobble_enabled=row.scrobble_enabled,
+        scrobble_provider=row.scrobble_provider,
+        scrobble_username=row.scrobble_username,
+        scrobble_session_key_enc=row.scrobble_session_key_enc,
+    )
+
+
+class SqlAlchemyUserSettingsRepository:
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def get(self, user_id: uuid.UUID) -> UserSettings | None:
+        row = await self._session.get(UserSettingsModel, user_id)
+        return _to_entity(row) if row is not None else None
+
+    async def upsert(self, settings: UserSettings) -> UserSettings:
+        """Create or replace the caller's settings row with ``settings`` in full.
+        The service merges partial updates against current values before calling
+        this, so the write always carries the complete desired state."""
+        row = await self._session.get(UserSettingsModel, settings.user_id)
+        if row is None:
+            row = UserSettingsModel(user_id=settings.user_id)
+            self._session.add(row)
+        row.theme = settings.theme
+        row.stream_quality = settings.stream_quality
+        row.scrobble_enabled = settings.scrobble_enabled
+        row.scrobble_provider = settings.scrobble_provider
+        row.scrobble_username = settings.scrobble_username
+        row.scrobble_session_key_enc = settings.scrobble_session_key_enc
+        await self._session.flush()
+        await self._session.refresh(row)
+        return _to_entity(row)

app/infrastructure/sources/registry.py

@@ -48,6 +48,10 @@ class SourceRegistry:
         """Every registered source that supports search (for cross-source search)."""
         return [cast(SearchableSource, b) for b in self._by_name.values() if hasattr(b, "search")]
 
+    def indexables(self) -> list[IndexableSource]:
+        """Every registered source that can be indexed (for a full re-scan)."""
+        return [cast(IndexableSource, b) for b in self._by_name.values() if hasattr(b, "scan")]
+
     def infos(self) -> list[SourceInfo]:
         return [backend.info() for backend in self._by_name.values()]
 

app/main.py

@@ -4,6 +4,7 @@ from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
 
 from app.api.errors import register_exception_handlers
 from app.api.health import router as health_router
@@ -40,6 +41,19 @@ def create_app() -> FastAPI:
     )
 
     app.add_middleware(CorrelationIdMiddleware)
+    # CORS added last → outermost, so browser preflight (OPTIONS) is answered
+    # before anything else. The web UI can connect cross-origin (direct :8000,
+    # a LAN IP, 127.0.0.1 vs localhost), which needs these headers. Bearer-token
+    # auth (no cookies) → wildcard origins are safe with allow_credentials=False.
+    if settings.cors_allow_origins:
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=settings.cors_allow_origins,
+            allow_credentials=False,
+            allow_methods=["*"],
+            allow_headers=["*"],
+            expose_headers=["Content-Range", "Accept-Ranges", "Content-Length"],
+        )
     register_exception_handlers(app)
 
     app.include_router(health_router)

app/workers/arq_worker.py

@@ -9,6 +9,7 @@ from arq.connections import RedisSettings
 
 from app.core.config import get_settings
 from app.core.logging import configure_logging, get_logger
+from app.workers.tasks.cleanup_task import cleanup_storage
 from app.workers.tasks.download_task import download_track
 from app.workers.tasks.enrich_task import enrich_track
 from app.workers.tasks.import_task import scan_local_folder
@@ -33,6 +34,7 @@ class WorkerSettings:
         enrich_track,
         download_track,
         materialize_track,
+        cleanup_storage,
     ]
     on_startup = startup
     on_shutdown = shutdown

app/workers/tasks/cleanup_task.py

@@ -0,0 +1,40 @@
+"""arq task: reconcile the catalogue against storage.
+
+Scans every *local* track and drops rows whose backing file has vanished from
+storage (a "dangling" reference). Remote placeholders have no local file yet and
+are skipped by the repository query. The filesystem checks (one ``exists()`` per
+track) are heavy, so this runs off the request cycle (CLAUDE.md). Guarded against
+a storage outage: if an implausibly large share of files look missing, it assumes
+the backend is down and aborts without deleting anything.
+"""
+
+from typing import Any
+
+from app.core.logging import get_logger
+from app.infrastructure.db import session_scope
+from app.infrastructure.db.repositories import SqlAlchemyTrackRepository
+from app.infrastructure.storage.provider import get_file_storage
+
+log = get_logger("worker.cleanup")
+
+# If more than this fraction of tracks look missing, assume the storage backend
+# is unavailable (not that the library really evaporated) and refuse to delete.
+_OUTAGE_GUARD_FRACTION = 0.5
+
+
+async def cleanup_storage(_ctx: dict[str, Any]) -> dict[str, Any]:
+    storage = get_file_storage()
+    async with session_scope() as session:
+        tracks = SqlAlchemyTrackRepository(session)
+        refs = await tracks.all_storage_refs()
+        missing = [track_id for track_id, uri in refs if not await storage.exists(uri)]
+
+        if refs and len(missing) > len(refs) * _OUTAGE_GUARD_FRACTION:
+            log.warning("cleanup_aborted_outage_guard", scanned=len(refs), missing=len(missing))
+            return {"scanned": len(refs), "removed": 0, "aborted": True}
+
+        for track_id in missing:
+            await tracks.delete(track_id)
+
+    log.info("cleanup_done", scanned=len(refs), removed=len(missing))
+    return {"scanned": len(refs), "removed": len(missing), "aborted": False}

tests/test_admin_api.py

@@ -0,0 +1,144 @@
+"""Integration tests for the admin instance-management endpoints.
+
+Covers ``/admin/services`` (dependency health), ``/admin/sources``,
+``/admin/reindex`` (enqueue), and ``/admin/settings`` (effective config) plus
+their admin gating. Requires a reachable Postgres; skips otherwise.
+"""
+
+import asyncio
+import os
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+from app.core.config import get_settings
+from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
+from app.infrastructure.db.repositories import (
+    SqlAlchemyRefreshTokenRepository,
+    SqlAlchemyUserRepository,
+)
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+_db_reachable_cache: bool | None = None
+
+
+async def _db_reachable() -> bool:
+    global _db_reachable_cache
+    if _db_reachable_cache is not None:
+        return _db_reachable_cache
+    from sqlalchemy import text
+
+    try:
+        async with asyncio.timeout(3):
+            async with get_engine().connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        _db_reachable_cache = True
+    except Exception:
+        _db_reachable_cache = False
+    return _db_reachable_cache
+
+
+@pytest.fixture
+async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
+    if not await _db_reachable():
+        pytest.skip("Postgres not reachable — integration test skipped.")
+
+    os.environ["MEDIA_PATH"] = str(tmp_path)
+    get_settings.cache_clear()
+
+    try:
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+
+        from app.application.user_service import UserService
+        from app.core.security import Argon2PasswordHasher
+
+        async with session_scope() as session:
+            svc = UserService(
+                users=SqlAlchemyUserRepository(session),
+                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
+                hasher=Argon2PasswordHasher(),
+            )
+            await svc.create_user(username="user", password="testpass1", is_superuser=False)
+            await svc.create_user(username="admin", password="testpass1", is_superuser=True)
+
+        from app.main import create_app
+
+        app = create_app()
+        async with LifespanManager(app):
+            transport = ASGITransport(app=app)
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                yield client
+
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await dispose_engine()
+    finally:
+        os.environ.pop("MEDIA_PATH", None)
+        get_settings.cache_clear()
+
+
+async def _auth(api: AsyncClient, username: str) -> dict[str, str]:
+    resp = await api.post(
+        "/api/v1/auth/login", json={"username": username, "password": "testpass1"}
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def test_services_reports_dependency_health(api: AsyncClient) -> None:
+    headers = await _auth(api, "admin")
+    resp = await api.get("/api/v1/admin/services", headers=headers)
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["database"] == "ok"
+    assert body["redis"] in ("ok", "down")  # redis is up in CI/dev, but don't hard-require it
+    assert body["ml"] == "skipped"  # no ML_SERVICE_URL configured
+
+
+async def test_services_requires_admin(api: AsyncClient) -> None:
+    headers = await _auth(api, "user")
+    resp = await api.get("/api/v1/admin/services", headers=headers)
+    assert resp.status_code == 403
+
+
+async def test_sources_list_is_admin_only(api: AsyncClient) -> None:
+    user = await _auth(api, "user")
+    assert (await api.get("/api/v1/admin/sources", headers=user)).status_code == 403
+
+    admin = await _auth(api, "admin")
+    resp = await api.get("/api/v1/admin/sources", headers=admin)
+    assert resp.status_code == 200, resp.text
+    assert isinstance(resp.json(), list)
+
+
+async def test_reindex_without_indexable_source_is_503(api: AsyncClient) -> None:
+    # No LOCAL_MEDIA_IMPORT_PATH configured → nothing to index.
+    headers = await _auth(api, "admin")
+    resp = await api.post("/api/v1/admin/reindex", headers=headers)
+    assert resp.status_code == 503, resp.text
+
+
+async def test_settings_exposes_effective_config_without_secrets(api: AsyncClient) -> None:
+    headers = await _auth(api, "admin")
+    resp = await api.get("/api/v1/admin/settings", headers=headers)
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    # environment reflects however the process booted (test on host, dev in the
+    # container) — assert it's a valid value, not a specific one.
+    assert body["environment"] in ("dev", "test", "prod")
+    assert body["storage_backend"] == "local"
+    assert body["allow_registration"] is True
+    # No secret material should ever appear in the payload.
+    assert "jwt_secret" not in body
+    assert "subsonic_secret_key" not in body
+
+
+async def test_settings_requires_admin(api: AsyncClient) -> None:
+    headers = await _auth(api, "user")
+    resp = await api.get("/api/v1/admin/settings", headers=headers)
+    assert resp.status_code == 403

tests/test_playlist_cover_api.py

@@ -0,0 +1,153 @@
+"""Integration test for the playlist cover endpoint.
+
+Master already covers playlist reorder in ``test_playlists_api.py``; this file
+only exercises ``GET /playlists/{id}/cover`` (served when set, 404 otherwise).
+Requires a reachable Postgres; skips otherwise.
+"""
+
+import asyncio
+import os
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+from app.core.config import get_settings
+from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
+from app.infrastructure.db.repositories import (
+    SqlAlchemyPlaylistRepository,
+    SqlAlchemyRefreshTokenRepository,
+    SqlAlchemyUserRepository,
+)
+from app.infrastructure.storage.provider import get_file_storage
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+# A minimal valid 1x1 PNG.
+_PNG_BYTES = bytes.fromhex(
+    "89504e470d0a1a0a0000000d4948445200000001000000010802000000907753"
+    "de0000000c4944415408d763f8cfc0f01f0005000155a2b4f60000000049454e44ae426082"
+)
+
+_db_reachable_cache: bool | None = None
+
+
+async def _db_reachable() -> bool:
+    global _db_reachable_cache
+    if _db_reachable_cache is not None:
+        return _db_reachable_cache
+    from sqlalchemy import text
+
+    try:
+        async with asyncio.timeout(3):
+            async with get_engine().connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        _db_reachable_cache = True
+    except Exception:
+        _db_reachable_cache = False
+    return _db_reachable_cache
+
+
+async def _seed_playlist(*, owner_id: uuid.UUID, cover: Path | None) -> uuid.UUID:
+    """Create a playlist owned by ``owner_id``; if ``cover`` is given, store it
+    and point the playlist's ``cover_path`` at it."""
+    from app.infrastructure.db.models.playlist import PlaylistModel
+
+    async with session_scope() as session:
+        playlist = await SqlAlchemyPlaylistRepository(session).add(
+            name="Mix", description=None, owner_id=owner_id
+        )
+        if cover is not None:
+            key = f"covers/playlists/{playlist.id}.png"
+            await get_file_storage().save_file(key, cover)
+            row = await session.get(PlaylistModel, playlist.id)
+            assert row is not None
+            row.cover_path = key
+        return playlist.id
+
+
+@pytest.fixture
+async def ctx(tmp_path: Path) -> AsyncIterator[tuple[AsyncClient, uuid.UUID, Path]]:
+    if not await _db_reachable():
+        pytest.skip("Postgres not reachable — integration test skipped.")
+
+    os.environ["MEDIA_PATH"] = str(tmp_path)
+    get_settings.cache_clear()
+
+    import app.infrastructure.storage.provider as _storage_provider
+
+    _storage_provider._storage = None
+
+    try:
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+
+        from app.application.user_service import UserService
+        from app.core.security import Argon2PasswordHasher
+
+        async with session_scope() as session:
+            user = await UserService(
+                users=SqlAlchemyUserRepository(session),
+                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
+                hasher=Argon2PasswordHasher(),
+            ).create_user(username="pluser", password="testpass1", is_superuser=False)
+            user_id = user.id
+
+        # A real source file for save_file (avoids the Windows NamedTemporaryFile
+        # re-open quirk).
+        src = tmp_path / "src_cover.png"
+        src.write_bytes(_PNG_BYTES)
+
+        from app.main import create_app
+
+        app = create_app()
+        async with LifespanManager(app):
+            transport = ASGITransport(app=app)
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                yield client, user_id, src
+
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await dispose_engine()
+    finally:
+        _storage_provider._storage = None
+        os.environ.pop("MEDIA_PATH", None)
+        get_settings.cache_clear()
+
+
+async def _token(api: AsyncClient) -> str:
+    resp = await api.post(
+        "/api/v1/auth/login", json={"username": "pluser", "password": "testpass1"}
+    )
+    assert resp.status_code == 200, resp.text
+    return str(resp.json()["access_token"])
+
+
+async def test_playlist_cover_served(ctx: tuple[AsyncClient, uuid.UUID, Path]) -> None:
+    api, user_id, src = ctx
+    token = await _token(api)
+    playlist_id = await _seed_playlist(owner_id=user_id, cover=src)
+
+    resp = await api.get(f"/api/v1/playlists/{playlist_id}/cover?token={token}")
+    assert resp.status_code == 200, resp.text
+    assert resp.headers["content-type"] == "image/png"
+    assert resp.content == _PNG_BYTES
+
+
+async def test_playlist_without_cover_is_404(ctx: tuple[AsyncClient, uuid.UUID, Path]) -> None:
+    api, user_id, _ = ctx
+    token = await _token(api)
+    playlist_id = await _seed_playlist(owner_id=user_id, cover=None)
+
+    resp = await api.get(f"/api/v1/playlists/{playlist_id}/cover?token={token}")
+    assert resp.status_code == 404
+
+
+async def test_playlist_cover_requires_auth(ctx: tuple[AsyncClient, uuid.UUID, Path]) -> None:
+    api, user_id, src = ctx
+    playlist_id = await _seed_playlist(owner_id=user_id, cover=src)
+    resp = await api.get(f"/api/v1/playlists/{playlist_id}/cover")
+    assert resp.status_code == 401

tests/test_sources_api.py

@@ -59,6 +59,13 @@ async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
     os.environ["LOCAL_MEDIA_IMPORT_PATH"] = str(music)
     get_settings.cache_clear()
 
+    # The source registry is process-cached (lru_cache); another test hitting a
+    # registry-backed endpoint (e.g. /admin/sources) may have built it before
+    # LOCAL_MEDIA_IMPORT_PATH was set. Clear it so this test sees a fresh one.
+    from app.api.deps import get_source_registry
+
+    get_source_registry.cache_clear()
+
     import app.infrastructure.storage.provider as _storage_provider
 
     _storage_provider._storage = None
@@ -94,6 +101,7 @@ async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
         os.environ.pop("MEDIA_PATH", None)
         os.environ.pop("LOCAL_MEDIA_IMPORT_PATH", None)
         get_settings.cache_clear()
+        get_source_registry.cache_clear()
 
 
 async def _login(api: AsyncClient) -> str:

tests/test_storage_maintenance_api.py

@@ -0,0 +1,185 @@
+"""Integration tests for the storage maintenance endpoints.
+
+Covers ``/storage/duplicates`` (shared fingerprint), ``/broken`` (failed
+enrichment), ``/missing-metadata`` (pending), and the admin-gated ``/cleanup``
+enqueue. Requires a reachable Postgres; skips otherwise.
+"""
+
+import asyncio
+import os
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+from app.core.config import get_settings
+from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
+from app.infrastructure.db.repositories import (
+    SqlAlchemyArtistRepository,
+    SqlAlchemyRefreshTokenRepository,
+    SqlAlchemyTrackRepository,
+    SqlAlchemyUserRepository,
+)
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+_db_reachable_cache: bool | None = None
+
+
+async def _db_reachable() -> bool:
+    global _db_reachable_cache
+    if _db_reachable_cache is not None:
+        return _db_reachable_cache
+    from sqlalchemy import text
+
+    try:
+        async with asyncio.timeout(3):
+            async with get_engine().connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        _db_reachable_cache = True
+    except Exception:
+        _db_reachable_cache = False
+    return _db_reachable_cache
+
+
+async def _seed_track(
+    *, title: str, source_id: str, metadata_status: str, fingerprint: str | None
+) -> uuid.UUID:
+    async with session_scope() as session:
+        artist = await SqlAlchemyArtistRepository(session).get_or_create("Storage Artist")
+        tracks = SqlAlchemyTrackRepository(session)
+        tid = uuid.uuid4()
+        await tracks.add(
+            id=tid,
+            title=title,
+            artist_id=artist.id,
+            storage_uri=f"tracks/zz/{source_id}.mp3",
+            file_format="mp3",
+            file_size=10,
+            source="upload",
+            source_id=source_id,
+            metadata_status="pending",
+            added_by=None,
+        )
+        # apply_enrichment lets us set fingerprint + final status precisely.
+        await tracks.apply_enrichment(
+            tid,
+            title=title,
+            artist_id=artist.id,
+            album_id=None,
+            genre=None,
+            year=None,
+            track_number=None,
+            duration_seconds=1,
+            bitrate=None,
+            acoustid_fingerprint=fingerprint,
+            musicbrainz_id=None,
+            metadata_status=metadata_status,
+        )
+        return tid
+
+
+@pytest.fixture
+async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
+    if not await _db_reachable():
+        pytest.skip("Postgres not reachable — integration test skipped.")
+
+    os.environ["MEDIA_PATH"] = str(tmp_path)
+    get_settings.cache_clear()
+
+    try:
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+
+        from app.application.user_service import UserService
+        from app.core.security import Argon2PasswordHasher
+
+        async with session_scope() as session:
+            svc = UserService(
+                users=SqlAlchemyUserRepository(session),
+                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
+                hasher=Argon2PasswordHasher(),
+            )
+            await svc.create_user(username="user", password="testpass1", is_superuser=False)
+            await svc.create_user(username="admin", password="testpass1", is_superuser=True)
+
+        from app.main import create_app
+
+        app = create_app()
+        async with LifespanManager(app):
+            transport = ASGITransport(app=app)
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                yield client
+
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await dispose_engine()
+    finally:
+        os.environ.pop("MEDIA_PATH", None)
+        get_settings.cache_clear()
+
+
+async def _auth(api: AsyncClient, username: str) -> dict[str, str]:
+    resp = await api.post(
+        "/api/v1/auth/login", json={"username": username, "password": "testpass1"}
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def test_duplicates_grouped_by_fingerprint(api: AsyncClient) -> None:
+    headers = await _auth(api, "user")
+    await _seed_track(title="Dup A", source_id="a", metadata_status="enriched", fingerprint="FP1")
+    await _seed_track(title="Dup B", source_id="b", metadata_status="enriched", fingerprint="FP1")
+    # A lone fingerprint must NOT show up as a duplicate.
+    await _seed_track(title="Solo", source_id="c", metadata_status="enriched", fingerprint="FP2")
+
+    resp = await api.get("/api/v1/storage/duplicates", headers=headers)
+    assert resp.status_code == 200, resp.text
+    groups = resp.json()
+    assert len(groups) == 1
+    assert groups[0]["fingerprint"] == "FP1"
+    assert len(groups[0]["tracks"]) == 2
+
+
+async def test_broken_lists_failed_tracks(api: AsyncClient) -> None:
+    headers = await _auth(api, "user")
+    await _seed_track(title="Bad", source_id="x", metadata_status="failed", fingerprint=None)
+    await _seed_track(title="Good", source_id="y", metadata_status="enriched", fingerprint=None)
+
+    resp = await api.get("/api/v1/storage/broken", headers=headers)
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["total"] == 1
+    assert body["items"][0]["title"] == "Bad"
+
+
+async def test_missing_metadata_lists_pending(api: AsyncClient) -> None:
+    headers = await _auth(api, "user")
+    await _seed_track(title="Pending", source_id="p", metadata_status="pending", fingerprint=None)
+    await _seed_track(title="Done", source_id="d", metadata_status="enriched", fingerprint=None)
+
+    resp = await api.get("/api/v1/storage/missing-metadata", headers=headers)
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["total"] == 1
+    assert body["items"][0]["title"] == "Pending"
+
+
+async def test_cleanup_requires_admin(api: AsyncClient) -> None:
+    user = await _auth(api, "user")
+    resp = await api.post("/api/v1/storage/cleanup", headers=user)
+    assert resp.status_code == 403
+
+
+async def test_cleanup_enqueues_for_admin(api: AsyncClient) -> None:
+    admin = await _auth(api, "admin")
+    resp = await api.post("/api/v1/storage/cleanup", headers=admin)
+    # 202 with a job id when the queue is reachable; 503 if Redis is down.
+    assert resp.status_code in (202, 503), resp.text
+    if resp.status_code == 202:
+        assert resp.json()["status"] == "enqueued"
+        assert resp.json()["job_id"]

tests/test_sync_api.py

@@ -0,0 +1,252 @@
+"""Integration tests for the offline-first sync endpoints.
+
+Drives POST /sync/push (idempotent append of like/play events) and GET
+/sync/changes (delta pull with a server-clock cursor). Requires a reachable
+Postgres; skips otherwise.
+"""
+
+import asyncio
+import os
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+from app.core.config import get_settings
+from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
+from app.infrastructure.db.repositories import (
+    SqlAlchemyArtistRepository,
+    SqlAlchemyRefreshTokenRepository,
+    SqlAlchemyTrackRepository,
+    SqlAlchemyUserRepository,
+)
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+_db_reachable_cache: bool | None = None
+
+
+async def _db_reachable() -> bool:
+    global _db_reachable_cache
+    if _db_reachable_cache is not None:
+        return _db_reachable_cache
+    from sqlalchemy import text
+
+    try:
+        async with asyncio.timeout(3):
+            async with get_engine().connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        _db_reachable_cache = True
+    except Exception:
+        _db_reachable_cache = False
+    return _db_reachable_cache
+
+
+async def _seed_track(source_id: str = "sync-1") -> uuid.UUID:
+    async with session_scope() as session:
+        artist = await SqlAlchemyArtistRepository(session).get_or_create("Sync Artist")
+        tid = uuid.uuid4()
+        await SqlAlchemyTrackRepository(session).add(
+            id=tid,
+            title="Sync Track",
+            artist_id=artist.id,
+            storage_uri=f"tracks/zz/{source_id}.mp3",
+            file_format="mp3",
+            file_size=10,
+            source="upload",
+            source_id=source_id,
+            metadata_status="enriched",
+            added_by=None,
+        )
+        return tid
+
+
+@pytest.fixture
+async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
+    if not await _db_reachable():
+        pytest.skip("Postgres not reachable — integration test skipped.")
+
+    os.environ["MEDIA_PATH"] = str(tmp_path)
+    get_settings.cache_clear()
+
+    try:
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+
+        from app.application.user_service import UserService
+        from app.core.security import Argon2PasswordHasher
+
+        async with session_scope() as session:
+            await UserService(
+                users=SqlAlchemyUserRepository(session),
+                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
+                hasher=Argon2PasswordHasher(),
+            ).create_user(username="syncer", password="testpass1", is_superuser=False)
+
+        from app.main import create_app
+
+        app = create_app()
+        async with LifespanManager(app):
+            transport = ASGITransport(app=app)
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                yield client
+
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await dispose_engine()
+    finally:
+        os.environ.pop("MEDIA_PATH", None)
+        get_settings.cache_clear()
+
+
+async def _auth(api: AsyncClient) -> dict[str, str]:
+    resp = await api.post(
+        "/api/v1/auth/login", json={"username": "syncer", "password": "testpass1"}
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def test_push_is_idempotent_and_pull_returns_events(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    track_id = await _seed_track()
+    like_id, play_id = str(uuid.uuid4()), str(uuid.uuid4())
+
+    push_body = {
+        "likes": [
+            {
+                "id": like_id,
+                "track_id": str(track_id),
+                "value": "like",
+                "created_at": "2020-01-01T00:00:00+00:00",
+            }
+        ],
+        "plays": [
+            {
+                "id": play_id,
+                "track_id": str(track_id),
+                "played_at": "2020-01-01T00:05:00+00:00",
+                "play_duration_seconds": 180,
+                "completed": True,
+            }
+        ],
+    }
+
+    first = await api.post("/api/v1/sync/push", json=push_body, headers=headers)
+    assert first.status_code == 200, first.text
+    assert first.json()["accepted_likes"] == 1
+    assert first.json()["accepted_plays"] == 1
+
+    # Replaying the exact same events must store nothing (idempotent by id).
+    replay = await api.post("/api/v1/sync/push", json=push_body, headers=headers)
+    assert replay.json()["accepted_likes"] == 0
+    assert replay.json()["accepted_plays"] == 0
+
+    # Full pull returns the events + the catalogue track, with a cursor.
+    changes = await api.get("/api/v1/sync/changes", headers=headers)
+    assert changes.status_code == 200, changes.text
+    body = changes.json()
+    assert [lk["id"] for lk in body["likes"]] == [like_id]
+    assert body["likes"][0]["created_at"].startswith("2020-01-01")  # event time preserved
+    assert [p["id"] for p in body["plays"]] == [play_id]
+    assert str(track_id) in [t["id"] for t in body["tracks"]]
+    assert body["cursor"]
+
+
+async def test_cursor_advances_and_empty_delta(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    track_id = await _seed_track()
+    await api.post(
+        "/api/v1/sync/push",
+        json={
+            "likes": [
+                {
+                    "id": str(uuid.uuid4()),
+                    "track_id": str(track_id),
+                    "value": "like",
+                    "created_at": "2024-01-01T00:00:00+00:00",
+                }
+            ]
+        },
+        headers=headers,
+    )
+    cursor = (await api.get("/api/v1/sync/changes", headers=headers)).json()["cursor"]
+
+    # Nothing changed since the cursor → empty delta (but a fresh cursor).
+    delta = await api.get("/api/v1/sync/changes", params={"since": cursor}, headers=headers)
+    assert delta.status_code == 200, delta.text
+    assert delta.json()["likes"] == []
+    assert delta.json()["plays"] == []
+
+
+async def test_offline_event_with_old_time_still_syncs(api: AsyncClient) -> None:
+    # The delta is keyed by server-ingestion time, not the client event time, so
+    # an event pushed now with an old created_at still surfaces for a cursor that
+    # sits *after* that old event time.
+    headers = await _auth(api)
+    track_id = await _seed_track()
+    baseline = (await api.get("/api/v1/sync/changes", headers=headers)).json()["cursor"]
+
+    await api.post(
+        "/api/v1/sync/push",
+        json={
+            "likes": [
+                {
+                    "id": str(uuid.uuid4()),
+                    "track_id": str(track_id),
+                    "value": "like",
+                    "created_at": "2019-06-01T00:00:00+00:00",  # long before `baseline`
+                }
+            ]
+        },
+        headers=headers,
+    )
+
+    delta = await api.get("/api/v1/sync/changes", params={"since": baseline}, headers=headers)
+    assert len(delta.json()["likes"]) == 1  # surfaced despite the old event time
+
+
+async def test_push_skips_unknown_track(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    resp = await api.post(
+        "/api/v1/sync/push",
+        json={
+            "likes": [
+                {
+                    "id": str(uuid.uuid4()),
+                    "track_id": str(uuid.uuid4()),  # not in the catalogue
+                    "value": "like",
+                    "created_at": "2024-01-01T00:00:00+00:00",
+                }
+            ]
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["accepted_likes"] == 0
+
+
+async def test_pull_includes_changed_playlists_with_track_ids(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    track_id = await _seed_track()
+
+    created = await api.post("/api/v1/playlists", json={"name": "Sync Mix"}, headers=headers)
+    playlist_id = created.json()["id"]
+    await api.post(
+        f"/api/v1/playlists/{playlist_id}/tracks",
+        json={"track_id": str(track_id)},
+        headers=headers,
+    )
+
+    body = (await api.get("/api/v1/sync/changes", headers=headers)).json()
+    playlists = {p["id"]: p for p in body["playlists"]}
+    assert playlist_id in playlists
+    assert playlists[playlist_id]["track_ids"] == [str(track_id)]
+
+
+async def test_sync_requires_auth(api: AsyncClient) -> None:
+    assert (await api.get("/api/v1/sync/changes")).status_code == 401
+    assert (await api.post("/api/v1/sync/push", json={})).status_code == 401

tests/test_user_settings_api.py

@@ -0,0 +1,174 @@
+"""Integration tests for the user-settings + scrobbling endpoints.
+
+Drives ``/api/v1/settings`` end to end: lazy defaults, partial update,
+enum validation, and scrobbling config (write-only session key). Requires a
+reachable Postgres; skips otherwise.
+"""
+
+import asyncio
+import os
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+from app.core.config import get_settings
+from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
+from app.infrastructure.db.repositories import (
+    SqlAlchemyRefreshTokenRepository,
+    SqlAlchemyUserRepository,
+)
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+_db_reachable_cache: bool | None = None
+
+
+async def _db_reachable() -> bool:
+    global _db_reachable_cache
+    if _db_reachable_cache is not None:
+        return _db_reachable_cache
+    from sqlalchemy import text
+
+    try:
+        async with asyncio.timeout(3):
+            async with get_engine().connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        _db_reachable_cache = True
+    except Exception:
+        _db_reachable_cache = False
+    return _db_reachable_cache
+
+
+@pytest.fixture
+async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
+    if not await _db_reachable():
+        pytest.skip("Postgres not reachable — integration test skipped.")
+
+    os.environ["MEDIA_PATH"] = str(tmp_path)
+    get_settings.cache_clear()
+
+    try:
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+
+        from app.application.user_service import UserService
+        from app.core.security import Argon2PasswordHasher
+
+        async with session_scope() as session:
+            await UserService(
+                users=SqlAlchemyUserRepository(session),
+                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
+                hasher=Argon2PasswordHasher(),
+            ).create_user(username="setuser", password="testpass1", is_superuser=False)
+
+        from app.main import create_app
+
+        app = create_app()
+        async with LifespanManager(app):
+            transport = ASGITransport(app=app)
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                yield client
+
+        async with get_engine().begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await dispose_engine()
+    finally:
+        os.environ.pop("MEDIA_PATH", None)
+        get_settings.cache_clear()
+
+
+async def _auth(api: AsyncClient) -> dict[str, str]:
+    resp = await api.post(
+        "/api/v1/auth/login", json={"username": "setuser", "password": "testpass1"}
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def test_get_settings_returns_defaults(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    resp = await api.get("/api/v1/settings", headers=headers)
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {"theme": "system", "stream_quality": "original"}
+
+
+async def test_patch_settings_persists(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    resp = await api.patch("/api/v1/settings", json={"theme": "dark"}, headers=headers)
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {"theme": "dark", "stream_quality": "original"}
+
+    # Persisted + partial update leaves the untouched field alone.
+    again = await api.patch("/api/v1/settings", json={"stream_quality": "low"}, headers=headers)
+    assert again.json() == {"theme": "dark", "stream_quality": "low"}
+
+
+async def test_invalid_theme_is_422(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    resp = await api.patch("/api/v1/settings", json={"theme": "neon"}, headers=headers)
+    assert resp.status_code == 422
+
+
+async def test_scrobbling_defaults_and_enable(api: AsyncClient) -> None:
+    headers = await _auth(api)
+
+    resp = await api.get("/api/v1/settings/scrobbling", headers=headers)
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {
+        "enabled": False,
+        "provider": None,
+        "username": None,
+        "configured": False,
+    }
+
+    # Enabling without a provider is rejected.
+    bad = await api.put("/api/v1/settings/scrobbling", json={"enabled": True}, headers=headers)
+    assert bad.status_code == 422
+
+    # Enable with a provider + secret token; the token is never echoed back.
+    ok = await api.put(
+        "/api/v1/settings/scrobbling",
+        json={
+            "enabled": True,
+            "provider": "listenbrainz",
+            "username": "me",
+            "session_key": "super-secret-token",
+        },
+        headers=headers,
+    )
+    assert ok.status_code == 200, ok.text
+    body = ok.json()
+    assert body == {
+        "enabled": True,
+        "provider": "listenbrainz",
+        "username": "me",
+        "configured": True,
+    }
+    assert "session_key" not in body
+    assert "super-secret-token" not in ok.text
+
+
+async def test_scrobbling_keeps_stored_key_when_omitted(api: AsyncClient) -> None:
+    headers = await _auth(api)
+    await api.put(
+        "/api/v1/settings/scrobbling",
+        json={"enabled": True, "provider": "lastfm", "session_key": "k"},
+        headers=headers,
+    )
+    # A later update without a session_key must keep the stored one.
+    resp = await api.put(
+        "/api/v1/settings/scrobbling",
+        json={"enabled": True, "provider": "lastfm", "username": "changed"},
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["configured"] is True
+    assert resp.json()["username"] == "changed"
+
+
+async def test_settings_require_auth(api: AsyncClient) -> None:
+    resp = await api.get("/api/v1/settings")
+    assert resp.status_code == 401