Blocklist Phase 1 (backend): artist/album/track bans enforced at the wishlist chokepoint

A proper artist/album/track blacklist (distinct from download_blacklist, which
stays untouched). ID-keyed across metadata sources so a ban survives a source
switch; profile-scoped; cascade artist→album→track.

- core/blocklist/matching.py — pure decision core (no I/O): build an index from
  rows, candidate_block_reason() walks track→album→artist. Same-source ID match
  is primary; artist NAME is a fallback (covers the ID-backfill window);
  albums/tracks are ID-only (common titles like "Greatest Hits" must not
  false-positive across artists). Source-isolated so a numeric Deezer id can't
  collide with a numeric iTunes id of a different entity.
- DB: new `blocklist` table (profile_id, entity_type, name, 4 source-id cols,
  match_status) + CRUD, match-row fetch, backfill-pending query, id-backfill
  update (COALESCE — fills NULLs only).
- Guard: _wishlist_blocklist_reason at the top of add_to_wishlist — every
  auto-acquisition path funnels through it, so one check covers watchlist,
  discography backfill, repair, manual add. Fails OPEN (a guard error never
  blocks a legitimate add).
- Discovery unified IN: legacy discovery_artist_blacklist is migrated into the
  blocklist on upgrade (replicated to every profile so no global ban silently
  stops working; idempotent; legacy table kept for rollback). Discovery reads
  (hero + personalized-playlist SQL) now union the blocklist, so a new-modal
  ban filters discovery too.

Tests: 13 on the pure matcher (cascade, id-vs-name rules, source isolation,
precedence) + 10 on the DB/guard (CRUD, profile isolation, dedup, backfill,
end-to-end wishlist refusal + cascade + the discovery migration upgrade path).
50 blocklist/personalized tests pass.

core/blocklist/__init__.py

@@ -0,0 +1,37 @@
+"""Artist / album / track blocklist (the "proper" blacklist).
+
+Distinct from ``download_blacklist`` (which skips one bad source file from one
+Soulseek peer — untouched here). This blocklist bans an ARTIST, ALBUM, or
+TRACK from being acquired, keyed by metadata-source IDs (Spotify / iTunes /
+Deezer / MusicBrainz) so a ban survives a source switch.
+
+Phase 1 enforces at the single ``add_to_wishlist`` chokepoint: every
+auto-acquisition path (watchlist, discography backfill, repair, manual
+wishlist add) funnels through it, so one guard covers them all.
+
+- ``matching`` — the pure decision core (no DB, no I/O): build an index from
+  blocklist rows, ask whether a candidate is blocked, with artist→album→track
+  cascade.
+"""
+
+from core.blocklist.matching import (
+    ENTITY_ALBUM,
+    ENTITY_ARTIST,
+    ENTITY_TRACK,
+    ENTITY_TYPES,
+    SOURCE_ID_FIELDS,
+    BlocklistIndex,
+    build_index,
+    candidate_block_reason,
+)
+
+__all__ = [
+    "ENTITY_ARTIST",
+    "ENTITY_ALBUM",
+    "ENTITY_TRACK",
+    "ENTITY_TYPES",
+    "SOURCE_ID_FIELDS",
+    "BlocklistIndex",
+    "build_index",
+    "candidate_block_reason",
+]

core/blocklist/matching.py

@@ -0,0 +1,128 @@
+"""Pure blocklist matching — no DB, no I/O, fully unit-testable.
+
+The brain of the blocklist: given the stored blocklist rows and a candidate
+track being considered for the wishlist, decide whether it's blocked.
+
+Design decisions (per Boulder):
+- **ID-keyed.** Each row carries the candidate's IDs in up to four metadata
+  sources. A candidate is matched against the SAME source it came in on
+  (the wishlist payload carries active-source IDs), so a Deezer-numeric id
+  can't collide with an iTunes-numeric id of a different entity.
+- **Cascade.** Blocking an artist blocks their albums + tracks; blocking an
+  album blocks its tracks. The candidate carries its own artist/album/track
+  IDs, so the check walks track → album → artist and blocks on the first hit.
+- **Name fallback for ARTISTS only.** A blocked artist also matches by
+  case-folded name — this covers the window before the background ID-backfill
+  has resolved the active source's id. Albums/tracks do NOT fall back to name
+  (common titles like "Greatest Hits" would false-positive across artists);
+  they rely on IDs, which backfill fills in.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+
+ENTITY_ARTIST = "artist"
+ENTITY_ALBUM = "album"
+ENTITY_TRACK = "track"
+ENTITY_TYPES = (ENTITY_ARTIST, ENTITY_ALBUM, ENTITY_TRACK)
+
+# Blocklist-row column → the metadata source it belongs to.
+SOURCE_ID_FIELDS = {
+    "spotify": "spotify_id",
+    "itunes": "itunes_id",
+    "deezer": "deezer_id",
+    "musicbrainz": "musicbrainz_id",
+}
+
+
+def _norm(text: Any) -> str:
+    return str(text or "").strip().casefold()
+
+
+@dataclass
+class _TypeIndex:
+    # per-source set of blocked ids, plus a case-folded name set (artists only)
+    ids: Dict[str, Set[str]] = field(default_factory=lambda: {s: set() for s in SOURCE_ID_FIELDS})
+    names: Set[str] = field(default_factory=set)
+
+    def hit(self, source: Optional[str], entity_id: Any, name: Any, use_name: bool) -> bool:
+        if entity_id and source in self.ids and str(entity_id) in self.ids[source]:
+            return True
+        if use_name and name and _norm(name) in self.names:
+            return True
+        return False
+
+
+@dataclass
+class BlocklistIndex:
+    """Membership index built once per scan from the blocklist rows."""
+    artists: _TypeIndex = field(default_factory=_TypeIndex)
+    albums: _TypeIndex = field(default_factory=_TypeIndex)
+    tracks: _TypeIndex = field(default_factory=_TypeIndex)
+
+    @property
+    def is_empty(self) -> bool:
+        for ti in (self.artists, self.albums, self.tracks):
+            if ti.names or any(ti.ids.values()):
+                return False
+        return True
+
+
+def build_index(rows: Iterable[Dict[str, Any]]) -> BlocklistIndex:
+    """Build a BlocklistIndex from blocklist DB rows.
+
+    Each row needs ``entity_type``, ``name``, and the source id columns
+    (``spotify_id`` / ``itunes_id`` / ``deezer_id`` / ``musicbrainz_id``).
+    Unknown entity types are ignored."""
+    idx = BlocklistIndex()
+    by_type = {ENTITY_ARTIST: idx.artists, ENTITY_ALBUM: idx.albums, ENTITY_TRACK: idx.tracks}
+    for row in rows or []:
+        ti = by_type.get((row.get("entity_type") or "").strip().lower())
+        if ti is None:
+            continue
+        for source, col in SOURCE_ID_FIELDS.items():
+            val = row.get(col)
+            if val:
+                ti.ids[source].add(str(val))
+        name = row.get("name")
+        if name:
+            ti.names.add(_norm(name))
+    return idx
+
+
+def candidate_block_reason(
+    index: BlocklistIndex,
+    *,
+    source: Optional[str],
+    track_id: Any = None,
+    track_name: Any = None,
+    album_id: Any = None,
+    album_name: Any = None,
+    artists: Optional[List[Dict[str, Any]]] = None,
+) -> Optional[Tuple[str, str]]:
+    """Return ``(entity_type, label)`` for the first cascade hit, else None.
+
+    ``source`` is the metadata source the candidate IDs came from (the wishlist
+    payload's provider). ``artists`` is a list of ``{'id', 'name'}`` dicts.
+    Order matters only for the returned reason — any hit blocks."""
+    if index.is_empty:
+        return None
+
+    # Track level — id only (names too ambiguous to ban across artists).
+    if index.tracks.hit(source, track_id, track_name, use_name=False):
+        return (ENTITY_TRACK, str(track_name or track_id or "track"))
+
+    # Album level — id only.
+    if index.albums.hit(source, album_id, album_name, use_name=False):
+        return (ENTITY_ALBUM, str(album_name or album_id or "album"))
+
+    # Artist level — id OR case-folded name (safe + covers the backfill window).
+    for artist in artists or []:
+        a_id = artist.get("id") if isinstance(artist, dict) else None
+        a_name = artist.get("name") if isinstance(artist, dict) else artist
+        if index.artists.hit(source, a_id, a_name, use_name=True):
+            return (ENTITY_ARTIST, str(a_name or a_id or "artist"))
+
+    return None

core/personalized_playlists.py

@@ -225,7 +225,7 @@ class PersonalizedPlaylistsService:
                 FROM discovery_pool
                 WHERE source = ?
                   AND (spotify_track_id IS NOT NULL OR itunes_track_id IS NOT NULL OR deezer_track_id IS NOT NULL)
-                  AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
+                  AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist UNION SELECT LOWER(name) FROM blocklist WHERE entity_type='artist')
                   {owned_clause}
                   {extra_where}
                 ORDER BY {order_by}
@@ -818,7 +818,7 @@ class PersonalizedPlaylistsService:
                         source
                     FROM discovery_pool
                     WHERE (artist_name LIKE ? OR track_name LIKE ?) AND source = ?
-                      AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
+                      AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist UNION SELECT LOWER(name) FROM blocklist WHERE entity_type='artist')
                     ORDER BY RANDOM()
                     LIMIT ?
                 """, (f'%{category}%', f'%{category}%', active_source, limit))

database/music_database.py

@@ -1930,6 +1930,31 @@ class MusicDatabase:
             cursor.execute("CREATE INDEX IF NOT EXISTS idx_dab_name ON discovery_artist_blacklist (artist_name COLLATE NOCASE)")
             cursor.execute("CREATE INDEX IF NOT EXISTS idx_dab_spotify ON discovery_artist_blacklist (spotify_artist_id)")
 
+            # Unified artist/album/track blocklist (the "proper" blacklist —
+            # distinct from download_blacklist, which is source-file skipping).
+            # ID-keyed across metadata sources so a ban survives a source
+            # switch; profile-scoped; enforced at add_to_wishlist. The old
+            # discovery_artist_blacklist is migrated in below.
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS blocklist (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    profile_id INTEGER NOT NULL DEFAULT 1,
+                    entity_type TEXT NOT NULL,        -- 'artist' | 'album' | 'track'
+                    name TEXT NOT NULL COLLATE NOCASE,
+                    spotify_id TEXT,
+                    itunes_id TEXT,
+                    deezer_id TEXT,
+                    musicbrainz_id TEXT,
+                    parent_name TEXT,                 -- display only (album's/track's artist)
+                    match_status TEXT DEFAULT 'pending',  -- pending | matched (cross-source backfill)
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            cursor.execute("CREATE INDEX IF NOT EXISTS idx_blocklist_profile_type ON blocklist (profile_id, entity_type)")
+            cursor.execute("CREATE INDEX IF NOT EXISTS idx_blocklist_spotify ON blocklist (spotify_id)")
+            cursor.execute("CREATE INDEX IF NOT EXISTS idx_blocklist_name ON blocklist (name COLLATE NOCASE)")
+            self._migrate_discovery_blacklist_into_blocklist(cursor)
+
             # Liked artists pool — aggregated followed/liked artists from connected services
             cursor.execute("""
                 CREATE TABLE IF NOT EXISTS liked_artists_pool (
@@ -8074,7 +8099,43 @@ class MusicDatabase:
         return presets.get(preset_name, presets["balanced"])
 
     # Wishlist management methods
-    
+
+    def _wishlist_blocklist_reason(self, profile_id, track_data):
+        """Return (entity_type, label) if this wishlist candidate is blocklisted,
+        else None. Pure matching lives in core.blocklist; this just pulls the
+        candidate's active-source IDs out of the payload and asks."""
+        try:
+            from core.blocklist import build_index, candidate_block_reason
+            rows = self.get_blocklist_rows_for_matching(profile_id)
+            if not rows:
+                return None
+            index = build_index(rows)
+            if index.is_empty:
+                return None
+            td = track_data or {}
+            source = (td.get('provider') or td.get('source') or '').strip().lower() or None
+            album = td.get('album') if isinstance(td.get('album'), dict) else {}
+            # Normalise artists to [{'id','name'}] from track + album credits.
+            artists = []
+            for a in (td.get('artists') or []):
+                if isinstance(a, dict):
+                    artists.append({'id': a.get('id'), 'name': a.get('name')})
+                elif a:
+                    artists.append({'id': None, 'name': str(a)})
+            for a in (album.get('artists') or []):
+                if isinstance(a, dict):
+                    artists.append({'id': a.get('id'), 'name': a.get('name')})
+            return candidate_block_reason(
+                index, source=source,
+                track_id=td.get('id'), track_name=td.get('name'),
+                album_id=album.get('id'), album_name=album.get('name'),
+                artists=artists,
+            )
+        except Exception as e:
+            # Never let the blocklist check break a wishlist add — fail open.
+            logger.debug("blocklist guard skipped: %s", e)
+            return None
+
     def add_to_wishlist(
         self,
         spotify_track_data: Dict[str, Any] = None,
@@ -8098,6 +8159,15 @@ class MusicDatabase:
                     logger.error("Cannot add track to wishlist: missing track ID")
                     return False
 
+                # Blocklist guard (Phase 1): every auto-acquisition path funnels
+                # through here, so one check blocks a banned artist/album/track
+                # (with artist→album→track cascade) before it can be queued.
+                _blocked = self._wishlist_blocklist_reason(profile_id, spotify_track_data)
+                if _blocked:
+                    logger.info("Skipping wishlist add — %s is blocklisted: '%s'",
+                                _blocked[0], _blocked[1])
+                    return False
+
                 from core.library import manual_library_match as _mlm
                 if _mlm.get_match_for_track(self, profile_id, spotify_track_data):
                     logger.info(
@@ -10998,16 +11068,197 @@ class MusicDatabase:
             return []
 
     def get_discovery_blacklist_names(self) -> set:
-        """Get set of blacklisted artist names (lowercased) for fast filtering."""
+        """Set of blacklisted artist names (lowercased) for discovery filtering.
+
+        Unions the legacy discovery_artist_blacklist with the new unified
+        blocklist's artist entries (across all profiles), so a ban added via
+        either path filters discovery. The legacy table is migrated into the
+        blocklist on upgrade but kept as a rollback safety net."""
         try:
             conn = self._get_connection()
             cursor = conn.cursor()
             cursor.execute("SELECT LOWER(artist_name) FROM discovery_artist_blacklist")
-            return {r[0] for r in cursor.fetchall()}
+            names = {r[0] for r in cursor.fetchall()}
+            try:
+                cursor.execute("SELECT LOWER(name) FROM blocklist WHERE entity_type = 'artist'")
+                names.update(r[0] for r in cursor.fetchall())
+            except Exception as _bl_err:  # noqa: BLE001 — old schema may predate blocklist
+                logger.debug("blocklist union skipped in discovery names: %s", _bl_err)
+            return names
         except Exception as e:
             logger.error(f"Error getting discovery blacklist names: {e}")
             return set()
 
+    # ==================== Blocklist (artist/album/track) ====================
+
+    def _migrate_discovery_blacklist_into_blocklist(self, cursor):
+        """One-time safe migration of the legacy global discovery blacklist into
+        the new profile-scoped blocklist as artist entries.
+
+        Replicated to EVERY existing profile so no existing discovery ban
+        silently stops working under the new per-profile model. Idempotent
+        (skips a (profile, name) already present). The old table is left in
+        place as a rollback safety net."""
+        try:
+            cursor.execute(
+                "SELECT artist_name, spotify_artist_id, itunes_artist_id, deezer_artist_id "
+                "FROM discovery_artist_blacklist")
+            legacy = cursor.fetchall()
+            if not legacy:
+                return
+            try:
+                cursor.execute("SELECT id FROM profiles")
+                profile_ids = [r[0] for r in cursor.fetchall()] or [1]
+            except Exception:
+                profile_ids = [1]
+
+            migrated = 0
+            for row in legacy:
+                name = row[0]
+                if not name:
+                    continue
+                for pid in profile_ids:
+                    cursor.execute(
+                        "SELECT 1 FROM blocklist WHERE profile_id = ? AND entity_type = 'artist' "
+                        "AND name = ? COLLATE NOCASE LIMIT 1", (pid, name))
+                    if cursor.fetchone():
+                        continue
+                    cursor.execute(
+                        "INSERT INTO blocklist (profile_id, entity_type, name, spotify_id, "
+                        "itunes_id, deezer_id, match_status) VALUES (?, 'artist', ?, ?, ?, ?, 'matched')",
+                        (pid, name, row[1], row[2], row[3]))
+                    migrated += 1
+            if migrated:
+                logger.info("Migrated %d discovery-blacklist artist entr(ies) into the "
+                            "unified blocklist across %d profile(s)", migrated, len(profile_ids))
+        except Exception as e:
+            logger.debug("discovery→blocklist migration skipped: %s", e)
+
+    def add_blocklist_entry(self, profile_id: int, entity_type: str, name: str,
+                            spotify_id: str = None, itunes_id: str = None,
+                            deezer_id: str = None, musicbrainz_id: str = None,
+                            parent_name: str = None) -> Optional[int]:
+        """Add an artist/album/track to the blocklist. Returns the new row id,
+        or an existing row's id if a matching (profile, type, id/name) is already
+        present. match_status starts 'pending' until the backfill resolves the
+        other sources (unless we already have multiple ids)."""
+        if entity_type not in ('artist', 'album', 'track') or not name:
+            return None
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            # Dedup: same profile+type with any overlapping source id, or same name.
+            cursor.execute(
+                """SELECT id FROM blocklist WHERE profile_id = ? AND entity_type = ?
+                   AND ( (spotify_id IS NOT NULL AND spotify_id = ?)
+                      OR (itunes_id IS NOT NULL AND itunes_id = ?)
+                      OR (deezer_id IS NOT NULL AND deezer_id = ?)
+                      OR (musicbrainz_id IS NOT NULL AND musicbrainz_id = ?)
+                      OR name = ? COLLATE NOCASE ) LIMIT 1""",
+                (profile_id, entity_type, spotify_id, itunes_id, deezer_id, musicbrainz_id, name))
+            existing = cursor.fetchone()
+            if existing:
+                return existing[0]
+            id_count = sum(1 for x in (spotify_id, itunes_id, deezer_id, musicbrainz_id) if x)
+            status = 'matched' if id_count >= 2 else 'pending'
+            cursor.execute(
+                """INSERT INTO blocklist (profile_id, entity_type, name, spotify_id, itunes_id,
+                   deezer_id, musicbrainz_id, parent_name, match_status)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (profile_id, entity_type, name, spotify_id, itunes_id, deezer_id,
+                 musicbrainz_id, parent_name, status))
+            conn.commit()
+            return cursor.lastrowid
+        except Exception as e:
+            logger.error(f"Error adding blocklist entry: {e}")
+            return None
+
+    def remove_blocklist_entry(self, profile_id: int, entry_id: int) -> bool:
+        """Remove a blocklist entry (scoped to the profile that owns it)."""
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.execute("DELETE FROM blocklist WHERE id = ? AND profile_id = ?",
+                           (int(entry_id), profile_id))
+            conn.commit()
+            return cursor.rowcount > 0
+        except Exception as e:
+            logger.error(f"Error removing blocklist entry: {e}")
+            return False
+
+    def get_blocklist(self, profile_id: int, entity_type: str = None) -> list:
+        """List blocklist entries for a profile, newest first."""
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            if entity_type:
+                cursor.execute(
+                    "SELECT id, profile_id, entity_type, name, spotify_id, itunes_id, deezer_id, "
+                    "musicbrainz_id, parent_name, match_status, created_at FROM blocklist "
+                    "WHERE profile_id = ? AND entity_type = ? ORDER BY created_at DESC",
+                    (profile_id, entity_type))
+            else:
+                cursor.execute(
+                    "SELECT id, profile_id, entity_type, name, spotify_id, itunes_id, deezer_id, "
+                    "musicbrainz_id, parent_name, match_status, created_at FROM blocklist "
+                    "WHERE profile_id = ? ORDER BY created_at DESC", (profile_id,))
+            return [dict(r) for r in cursor.fetchall()]
+        except Exception as e:
+            logger.error(f"Error getting blocklist: {e}")
+            return []
+
+    def get_blocklist_rows_for_matching(self, profile_id: int) -> list:
+        """Lightweight rows (entity_type + id columns + name) for building the
+        in-memory match index — used by the add_to_wishlist guard."""
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT entity_type, name, spotify_id, itunes_id, deezer_id, musicbrainz_id "
+                "FROM blocklist WHERE profile_id = ?", (profile_id,))
+            return [dict(r) for r in cursor.fetchall()]
+        except Exception as e:
+            logger.error(f"Error getting blocklist match rows: {e}")
+            return []
+
+    def get_blocklist_entries_needing_backfill(self) -> list:
+        """Entries still 'pending' cross-source ID resolution."""
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT id, profile_id, entity_type, name, spotify_id, itunes_id, deezer_id, "
+                "musicbrainz_id, parent_name FROM blocklist WHERE match_status = 'pending'")
+            return [dict(r) for r in cursor.fetchall()]
+        except Exception as e:
+            logger.error(f"Error getting blocklist backfill entries: {e}")
+            return []
+
+    def update_blocklist_entry_ids(self, entry_id: int, *, spotify_id: str = None,
+                                   itunes_id: str = None, deezer_id: str = None,
+                                   musicbrainz_id: str = None, mark_matched: bool = True) -> bool:
+        """Backfill resolved source IDs onto an entry (only fills NULLs)."""
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            sets, params = [], []
+            for col, val in (("spotify_id", spotify_id), ("itunes_id", itunes_id),
+                             ("deezer_id", deezer_id), ("musicbrainz_id", musicbrainz_id)):
+                if val:
+                    sets.append(f"{col} = COALESCE({col}, ?)")
+                    params.append(val)
+            if mark_matched:
+                sets.append("match_status = 'matched'")
+            if not sets:
+                return False
+            params.append(int(entry_id))
+            cursor.execute(f"UPDATE blocklist SET {', '.join(sets)} WHERE id = ?", params)
+            conn.commit()
+            return cursor.rowcount > 0
+        except Exception as e:
+            logger.error(f"Error updating blocklist entry ids: {e}")
+            return False
+
     # ==================== Liked Artists Pool Methods ====================
 
     @staticmethod

tests/blocklist/test_blocklist_db.py

@@ -0,0 +1,145 @@
+"""Blocklist DB layer + the add_to_wishlist enforcement guard.
+
+Verified against a real SQLite DB: CRUD + profile scoping + dedup + backfill
+update, the discovery→blocklist migration, and the end-to-end wishlist guard
+(a banned artist's track is refused; cascade; profile isolation).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from database.music_database import MusicDatabase
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return MusicDatabase(str(tmp_path / "m.db"))
+
+
+def _track(track_id, name, artist_id, artist_name, album_id="al0", album_name="Al", source="spotify"):
+    return {
+        "id": track_id, "name": name, "source": source,
+        "artists": [{"id": artist_id, "name": artist_name}],
+        "album": {"id": album_id, "name": album_name, "artists": [{"id": artist_id, "name": artist_name}]},
+    }
+
+
+# ── CRUD + scoping ───────────────────────────────────────────────────────────
+
+def test_add_list_remove(db):
+    eid = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
+    assert eid
+    rows = db.get_blocklist(1)
+    assert len(rows) == 1 and rows[0]["name"] == "Drake"
+    assert db.remove_blocklist_entry(1, eid) is True
+    assert db.get_blocklist(1) == []
+
+
+def test_dedup_by_id_and_name(db):
+    a = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")
+    b = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")   # same id
+    c = db.add_blocklist_entry(1, "artist", "drake")                     # same name (NOCASE)
+    assert a == b == c
+    assert len(db.get_blocklist(1)) == 1
+
+
+def test_profile_isolation(db):
+    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")
+    assert len(db.get_blocklist(1)) == 1
+    assert db.get_blocklist(2) == []
+    # remove is profile-scoped — profile 2 can't delete profile 1's row
+    eid = db.get_blocklist(1)[0]["id"]
+    assert db.remove_blocklist_entry(2, eid) is False
+    assert len(db.get_blocklist(1)) == 1
+
+
+def test_match_status_pending_vs_matched(db):
+    one = db.add_blocklist_entry(1, "artist", "A", spotify_id="sp")
+    two = db.add_blocklist_entry(1, "artist", "B", spotify_id="sp2", deezer_id="dz2")
+    rows = {r["id"]: r for r in db.get_blocklist(1)}
+    assert rows[one]["match_status"] == "pending"   # single id → needs backfill
+    assert rows[two]["match_status"] == "matched"   # 2+ ids → already cross-known
+
+
+def test_backfill_update_fills_only_nulls(db):
+    eid = db.add_blocklist_entry(1, "artist", "A", spotify_id="sp")
+    db.update_blocklist_entry_ids(eid, deezer_id="dz", spotify_id="SHOULD-NOT-OVERWRITE")
+    row = db.get_blocklist(1)[0]
+    assert row["spotify_id"] == "sp"        # COALESCE: existing kept
+    assert row["deezer_id"] == "dz"
+    assert row["match_status"] == "matched"
+    assert db.get_blocklist_entries_needing_backfill() == []
+
+
+# ── the wishlist guard, end to end ───────────────────────────────────────────
+
+def test_blocked_artist_track_refused_from_wishlist(db):
+    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
+    ok = db.add_to_wishlist(
+        spotify_track_data=_track("t1", "God's Plan", "drake-sp", "Drake"),
+        profile_id=1)
+    assert ok is False                       # refused
+    # And nothing landed in the wishlist.
+    assert db.get_wishlist_count(profile_id=1) == 0
+
+
+def test_unblocked_artist_track_is_added(db):
+    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
+    ok = db.add_to_wishlist(
+        spotify_track_data=_track("t2", "Hello", "adele-sp", "Adele"),
+        profile_id=1)
+    assert ok is True
+    assert db.get_wishlist_count(profile_id=1) == 1
+
+
+def test_block_is_profile_scoped_at_the_guard(db):
+    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
+    # Profile 2 has no such ban → the same track is allowed.
+    ok = db.add_to_wishlist(
+        spotify_track_data=_track("t3", "Nice For What", "drake-sp", "Drake"),
+        profile_id=2)
+    assert ok is True
+
+
+def test_album_block_cascades_to_track_at_guard(db):
+    db.add_blocklist_entry(1, "album", "Scorpion", spotify_id="scorp-sp")
+    ok = db.add_to_wishlist(
+        spotify_track_data=_track("t4", "Nonstop", "drake-sp", "Drake",
+                                  album_id="scorp-sp", album_name="Scorpion"),
+        profile_id=1)
+    assert ok is False
+
+
+def test_discovery_blacklist_migrated_into_blocklist(tmp_path):
+    import database.music_database as mdb
+
+    def _reinit(path):
+        # An app upgrade = a fresh process with an empty init memo. Clear the
+        # per-process "already initialized" set so init (and the migration)
+        # actually re-runs against the existing DB file.
+        mdb._database_initialized_paths.discard(str(mdb.Path(path).resolve()))
+        return MusicDatabase(path)
+
+    path = str(tmp_path / "mig.db")
+    db = MusicDatabase(path)
+    conn = db._get_connection()
+    cur = conn.cursor()
+    cur.execute("INSERT INTO discovery_artist_blacklist (artist_name, spotify_artist_id) "
+                "VALUES ('Nickelback', 'nb-sp')")
+    conn.commit()
+    conn.close()
+
+    db2 = _reinit(path)          # upgrade → migration runs (committed)
+    rows = db2.get_blocklist(1)
+    assert any(r["name"] == "Nickelback" and r["entity_type"] == "artist" for r in rows)
+
+    db3 = _reinit(path)          # idempotent — a second upgrade doesn't duplicate
+    nb = [r for r in db3.get_blocklist(1) if r["name"] == "Nickelback"]
+    assert len(nb) == 1
+
+    # The migrated ban actually enforces at the wishlist guard.
+    ok = db3.add_to_wishlist(
+        spotify_track_data=_track("t5", "Photograph", "nb-sp", "Nickelback"),
+        profile_id=1)
+    assert ok is False

tests/blocklist/test_matching.py

@@ -0,0 +1,130 @@
+"""Pure blocklist matching — the cascade + ID/name rules.
+
+Block an artist/album/track by metadata-source ID; a candidate track is
+blocked if its track, album, or any artist matches (cascade). Same-source ID
+match is primary; artist NAME is a fallback (covers the backfill window);
+albums/tracks are ID-only to avoid common-title false positives.
+"""
+
+from __future__ import annotations
+
+from core.blocklist.matching import (
+    ENTITY_ALBUM,
+    ENTITY_ARTIST,
+    ENTITY_TRACK,
+    build_index,
+    candidate_block_reason,
+)
+
+
+def _row(entity_type, name="", **ids):
+    return {"entity_type": entity_type, "name": name, **ids}
+
+
+def _check(index, source="spotify", **kw):
+    return candidate_block_reason(index, source=source, **kw)
+
+
+# ── empty / no-match ─────────────────────────────────────────────────────────
+
+def test_empty_blocklist_blocks_nothing():
+    idx = build_index([])
+    assert idx.is_empty
+    assert _check(idx, track_id="t1", album_id="al1",
+                  artists=[{"id": "ar1", "name": "X"}]) is None
+
+
+def test_unrelated_candidate_not_blocked():
+    idx = build_index([_row(ENTITY_ARTIST, "Drake", spotify_id="drake-sp")])
+    assert _check(idx, artists=[{"id": "other", "name": "Adele"}]) is None
+
+
+# ── artist level + cascade ───────────────────────────────────────────────────
+
+def test_artist_blocked_by_id_blocks_their_track():
+    idx = build_index([_row(ENTITY_ARTIST, "Drake", spotify_id="drake-sp")])
+    reason = _check(idx, track_id="t9", album_id="al9",
+                    artists=[{"id": "drake-sp", "name": "Drake"}])
+    assert reason == (ENTITY_ARTIST, "Drake")
+
+
+def test_artist_blocked_by_name_fallback():
+    # No id resolved for this source yet (backfill window) — name still catches it.
+    idx = build_index([_row(ENTITY_ARTIST, "Drake", deezer_id="drake-dz")])
+    reason = _check(idx, source="spotify", artists=[{"id": "drake-sp", "name": "drake"}])
+    assert reason == (ENTITY_ARTIST, "drake")
+
+
+def test_artist_name_match_is_case_insensitive():
+    idx = build_index([_row(ENTITY_ARTIST, "Tyler, The Creator", spotify_id="x")])
+    assert _check(idx, artists=[{"id": None, "name": "tyler, the creator"}]) is not None
+
+
+# ── album level ──────────────────────────────────────────────────────────────
+
+def test_album_blocked_by_id_blocks_its_track():
+    idx = build_index([_row(ENTITY_ALBUM, "Scorpion", spotify_id="scorp-sp")])
+    reason = _check(idx, track_id="t1", album_id="scorp-sp", album_name="Scorpion",
+                    artists=[{"id": "drake-sp", "name": "Drake"}])
+    assert reason == (ENTITY_ALBUM, "Scorpion")
+
+
+def test_album_name_does_not_match_without_id():
+    # Common title must NOT block across artists on name alone.
+    idx = build_index([_row(ENTITY_ALBUM, "Greatest Hits", spotify_id="gh-queen")])
+    reason = _check(idx, album_id="gh-abba", album_name="Greatest Hits",
+                    artists=[{"id": "abba", "name": "ABBA"}])
+    assert reason is None
+
+
+# ── track level ──────────────────────────────────────────────────────────────
+
+def test_track_blocked_by_id():
+    idx = build_index([_row(ENTITY_TRACK, "Hotline Bling", spotify_id="hb-sp")])
+    reason = _check(idx, track_id="hb-sp", track_name="Hotline Bling",
+                    album_id="al", artists=[{"id": "drake", "name": "Drake"}])
+    assert reason == (ENTITY_TRACK, "Hotline Bling")
+
+
+def test_track_name_alone_does_not_block():
+    idx = build_index([_row(ENTITY_TRACK, "Intro", spotify_id="intro-1")])
+    assert _check(idx, track_id="intro-2", track_name="Intro",
+                  artists=[{"id": "z", "name": "Z"}]) is None
+
+
+# ── source isolation (numeric id collision guard) ────────────────────────────
+
+def test_same_id_different_source_does_not_collide():
+    # Deezer artist 12246 is blocked; an iTunes artist that happens to be 12246
+    # is a DIFFERENT entity and must NOT match (candidate source = itunes).
+    idx = build_index([_row(ENTITY_ARTIST, "Some Deezer Artist", deezer_id="12246")])
+    reason = _check(idx, source="itunes", artists=[{"id": "12246", "name": "Other"}])
+    assert reason is None
+
+
+def test_same_id_same_source_matches():
+    idx = build_index([_row(ENTITY_ARTIST, "A", deezer_id="12246")])
+    reason = _check(idx, source="deezer", artists=[{"id": "12246", "name": "A"}])
+    assert reason is not None
+
+
+# ── multi-source row ─────────────────────────────────────────────────────────
+
+def test_row_with_multiple_source_ids_matches_each():
+    idx = build_index([_row(ENTITY_ARTIST, "Drake",
+                            spotify_id="sp", itunes_id="it", deezer_id="dz")])
+    assert _check(idx, source="spotify", artists=[{"id": "sp", "name": "Drake"}])
+    assert _check(idx, source="itunes", artists=[{"id": "it", "name": "Drake"}])
+    assert _check(idx, source="deezer", artists=[{"id": "dz", "name": "Drake"}])
+
+
+# ── cascade precedence ───────────────────────────────────────────────────────
+
+def test_track_hit_reported_before_artist():
+    idx = build_index([
+        _row(ENTITY_ARTIST, "Drake", spotify_id="drake"),
+        _row(ENTITY_TRACK, "God's Plan", spotify_id="gp"),
+    ])
+    reason = _check(idx, track_id="gp", track_name="God's Plan",
+                    artists=[{"id": "drake", "name": "Drake"}])
+    assert reason[0] == ENTITY_TRACK   # most specific hit wins the reason

tests/test_personalized_playlists_id_gate.py

@@ -68,6 +68,13 @@ class _FakeDatabase:
             CREATE TABLE discovery_artist_blacklist (
                 artist_name TEXT PRIMARY KEY
             );
+            -- Unified blocklist — discovery filtering now unions artist bans
+            -- from here too (Phase 1 blocklist). Minimal shape for the subquery.
+            CREATE TABLE blocklist (
+                id INTEGER PRIMARY KEY,
+                entity_type TEXT,
+                name TEXT
+            );
             -- Minimal `tracks` table: exists so the `exclude_owned`
             -- subquery in `_select_discovery_tracks` can join. Real
             -- schema has many more columns; we only need the source-id