music-assistant-server

1"""Base (ABC) MediaType specific controller."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from abc import ABCMeta, abstractmethod
8from collections.abc import Iterable
9from contextlib import suppress
10from contextvars import ContextVar
11from dataclasses import dataclass
12from datetime import UTC, datetime
13from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, final, overload
14
15from music_assistant_models.auth import Scope
16from music_assistant_models.enums import (
17    EventType,
18    ExternalID,
19    ImageType,
20    MediaType,
21    ProviderFeature,
22    ProviderType,
23)
24from music_assistant_models.errors import (
25    InsufficientPermissions,
26    InvalidDataError,
27    MediaNotFoundError,
28    ProviderUnavailableError,
29)
30from music_assistant_models.helpers import create_safe_string, get_global_cache_value
31from music_assistant_models.media_items import (
32    AudioFormat,
33    ItemMapping,
34    ItemMappingSummary,
35    MediaCollection,
36    MediaItemImage,
37    MediaItemMetadata,
38    MediaItemMetadataSummary,
39    MediaItemSummaryType,
40    MediaItemType,
41    ProviderMapping,
42    UniqueList,
43)
44
45from music_assistant.constants import (
46    DB_TABLE_ALBUM_ARTISTS,
47    DB_TABLE_ALBUM_TRACKS,
48    DB_TABLE_AUDIO_ANALYSIS,
49    DB_TABLE_AUDIOBOOK_ARTISTS,
50    DB_TABLE_EXTERNAL_ID_LOOKUP,
51    DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
52    DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
53    DB_TABLE_PLAYLOG,
54    DB_TABLE_PROVIDER_MAPPINGS,
55    DB_TABLE_TRACK_ARTISTS,
56    MASS_LOGGER_NAME,
57)
58from music_assistant.controllers.music.helpers import search_name_match_clause
59from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
60from music_assistant.helpers.collections import (
61    get_collection_item_id,
62    get_collection_name_from_item_id,
63)
64from music_assistant.helpers.compare import compare_media_item
65from music_assistant.helpers.database import UNSET
66from music_assistant.helpers.external_ids import (
67    external_id_lookup_values,
68    external_id_lookup_values_untyped,
69    external_id_sort_key,
70    normalize_external_ids,
71)
72from music_assistant.helpers.json import json_loads, serialize_to_json
73from music_assistant.helpers.util import guard_single_request, parse_optional_bool
74
75if TYPE_CHECKING:
76    from collections.abc import AsyncGenerator, Mapping
77
78    from music_assistant import MusicAssistant
79    from music_assistant.models.music_provider import MusicProvider
80    from music_assistant.models.plugin import PluginProvider
81
82
83ItemCls = TypeVar("ItemCls", bound="MediaItemType")
84
85
86JSON_KEYS = (
87    "artists",
88    "track_album",
89    "metadata",
90    "provider_mappings",
91    "external_ids",
92    "narrators",
93    "authors",
94    "genre_aliases",
95    "supported_mediatypes",
96    "translation_params",
97    "audiobook_artists",
98)
99
100# The columns that make up a relation row, so a merge can copy it onto the target
101# without relying on SELECT *: album_tracks carries a surrogate autoincrement id that
102# must not be copied along.
103RELATION_TABLE_COLUMNS = {
104    DB_TABLE_ALBUM_ARTISTS: ("album_id", "artist_id"),
105    DB_TABLE_ALBUM_TRACKS: ("track_id", "album_id", "disc_number", "track_number"),
106    DB_TABLE_AUDIOBOOK_ARTISTS: ("audiobook_id", "artist_id"),
107    DB_TABLE_TRACK_ARTISTS: ("track_id", "artist_id"),
108}
109
110# When set (task-local), per-item MEDIA_ITEM_ADDED/UPDATED events and the on_item_updated
111# provider write-back are suppressed, so bulk operations (provider sync, provider cleanup)
112# don't flood subscribers with one event per touched item.
113SUPPRESS_MEDIA_ITEM_UPDATES: ContextVar[bool] = ContextVar(
114    "SUPPRESS_MEDIA_ITEM_UPDATES", default=False
115)
116
117SORT_KEYS = {
118    # sqlite has no builtin support for natural sorting
119    # so we have use an additional column for this
120    # this also improves searching and sorting performance
121    "name": "search_name ASC",
122    "name_desc": "search_name DESC",
123    "duration": "duration ASC",
124    "duration_desc": "duration DESC",
125    "sort_name": "search_sort_name ASC",
126    "sort_name_desc": "search_sort_name DESC",
127    "timestamp_added": "timestamp_added ASC",
128    "timestamp_added_desc": "timestamp_added DESC",
129    "timestamp_modified": "timestamp_modified ASC",
130    "timestamp_modified_desc": "timestamp_modified DESC",
131    "last_played": "last_played ASC",
132    "last_played_desc": "last_played DESC",
133    "play_count": "play_count ASC",
134    "play_count_desc": "play_count DESC",
135    "year": "year ASC",
136    "year_desc": "year DESC",
137    "position": "position ASC",
138    "position_desc": "position DESC",
139    "album_artist_name": "artists.search_name ASC, year DESC",
140    "album_artist_name_desc": "artists.search_name DESC, year DESC",
141    "track_artist_name": "artists.search_name ASC, search_name ASC",
142    "track_artist_name_desc": "artists.search_name DESC, search_name ASC",
143    "random": "RANDOM()",
144    "random_play_count": "RANDOM(), play_count ASC",
145}
146
147
148@dataclass(slots=True)
149class LibraryItemSyncDetails:
150    """
151    Lightweight snapshot of a library item with just the fields the library sync needs.
152
153    Used by the provider sync loops to detect (un)changed items without hydrating
154    full MediaItem objects from the database.
155    """
156
157    item_id: int
158    favorite: bool
159    date_added: datetime
160    provider_mappings: set[ProviderMapping]
161
162
163@dataclass(slots=True)
164class TrackSyncDetails(LibraryItemSyncDetails):
165    """Lightweight sync snapshot of a library track."""
166
167    has_album: bool
168    has_artists: bool
169
170
171@dataclass(slots=True)
172class AudiobookSyncDetails(LibraryItemSyncDetails):
173    """Lightweight sync snapshot of a library audiobook."""
174
175    author_is_str: bool
176    narrator_is_str: bool
177    fully_played: bool | None
178    resume_position_ms: int | None
179
180
181class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta):
182    """Base model for controller managing a MediaType."""
183
184    media_type: MediaType
185    item_cls: type[MediaItemType]
186    summary_item_cls: type[MediaItemSummaryType]
187    db_table: str
188
189    def __init__(self, mass: MusicAssistant) -> None:
190        """Initialize class."""
191        self.mass = mass
192        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.music.{self.media_type.value}")
193        # register (base) api handlers
194        self.api_base = api_base = f"{self.media_type}s"
195        self.mass.register_api_command(
196            f"music/{api_base}/count", self.library_count, required_scope=Scope.LIBRARY_READ
197        )
198        self.mass.register_api_command(
199            f"music/{api_base}/library_items",
200            self.library_items,
201            required_scope=Scope.LIBRARY_READ,
202            allow_impersonation=True,
203        )
204        self.mass.register_api_command(
205            f"music/{api_base}/get", self.get, required_scope=Scope.LIBRARY_READ
206        )
207        self.mass.register_api_command(
208            f"music/{api_base}/get_by_external_id",
209            self.get_library_item_by_external_id,
210            required_scope=Scope.LIBRARY_READ,
211        )
212        self.mass.register_api_command(
213            f"music/{api_base}/get_collection",
214            self.get_collection,
215            required_scope=Scope.LIBRARY_READ,
216            allow_impersonation=True,
217        )
218        # Backward compatibility alias - prefer the generic "get" endpoint
219        self.mass.register_api_command(
220            f"music/{api_base}/get_{self.media_type}",
221            self.get,
222            required_scope=Scope.LIBRARY_READ,
223            alias=True,
224        )
225        self.mass.register_api_command(
226            f"music/{api_base}/update",
227            self.update_item_in_library,
228            required_scope=Scope.LIBRARY_MANAGE,
229        )
230        self.mass.register_api_command(
231            f"music/{api_base}/remove",
232            self.remove_item_from_library,
233            required_scope=Scope.LIBRARY_MANAGE,
234        )
235        self._db_add_lock = asyncio.Lock()
236
237    @property
238    def translation_owner(self) -> str:
239        """Return the "core.music" namespace these media controllers' translation strings live under."""
240        return "core.music"
241
242    @property
243    def base_query(self) -> tuple[str, dict[str, Any]]:
244        """
245        Return the base SELECT query for this media type and its bound query params.
246
247        Override in a subclass to customize the query (extra joins/columns) and/or to
248        inject dynamic, parameterized filters.
249        """
250        query = f"""
251        SELECT
252            {self.db_table}.*,
253            {self._external_ids_query()} AS external_ids,
254            {self._provider_mappings_query()} AS provider_mappings
255            FROM {self.db_table} """
256        return query, {}
257
258    @property
259    def summary_query(self) -> tuple[str, dict[str, Any]]:
260        """
261        Return the slim SELECT query used for summary listings and its bound query params.
262
263        Selects only the columns needed to build summary items. Override in a subclass
264        to select additional per-type columns.
265        """
266        query = f"""
267        SELECT
268            {self._summary_base_columns()},
269            {self._provider_mappings_query()} AS provider_mappings
270            FROM {self.db_table} """
271        return query, {}
272
273    @final
274    async def add_item_to_library(
275        self,
276        item: ItemCls,
277        overwrite_existing: bool = False,
278    ) -> ItemCls:
279        """Add item to library and return the new (or updated) database item."""
280        new_item = False
281        # batch the many writes of an item add/update into a single commit
282        async with self.mass.music.database.deferred_commit():
283            # check for existing item first
284            if library_id := await self._get_library_item_by_match(item):
285                # update existing item
286                await self._update_library_item(library_id, item, overwrite=overwrite_existing)
287            else:
288                # actually add a new item in the library db
289                self.mass.music.match_provider_instances(item)
290                async with self._db_add_lock:
291                    # Another task may have inserted the same item while this task waited.
292                    if library_id := await self._get_library_item_by_match(item):
293                        await self._update_library_item(
294                            library_id, item, overwrite=overwrite_existing
295                        )
296                    else:
297                        library_id = await self._add_library_item(item)
298                        new_item = True
299        # return final library_item
300        library_item = await self.get_library_item(library_id)
301        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
302            self.mass.signal_event(
303                EventType.MEDIA_ITEM_ADDED if new_item else EventType.MEDIA_ITEM_UPDATED,
304                library_item.uri,
305                library_item,
306            )
307        return library_item
308
309    @final
310    async def update_item_in_library(
311        self, item_id: str | int, update: ItemCls, overwrite: bool = False
312    ) -> ItemCls:
313        """Update existing library record in the library database."""
314        self.mass.music.match_provider_instances(update)
315        # batch the many writes of an item update into a single commit
316        async with self.mass.music.database.deferred_commit():
317            await self._update_library_item(item_id, update, overwrite=overwrite)
318        # return the updated object
319        library_item = await self.get_library_item(item_id)
320        if SUPPRESS_MEDIA_ITEM_UPDATES.get():
321            # during a sync the update originates from the provider itself,
322            # so skip both the event and the write-back to that provider
323            return library_item
324        # drop cached artwork for the updated item so replaced art is served fresh
325        for img in library_item.metadata.images or []:
326            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
327        self.mass.signal_event(
328            EventType.MEDIA_ITEM_UPDATED,
329            library_item.uri,
330            library_item,
331        )
332        # notify music providers of the update so they can sync their own storage
333        for prov_mapping in library_item.provider_mappings:
334            if provider := self.mass.get_provider(prov_mapping.provider_instance):
335                if provider.type != ProviderType.MUSIC:
336                    continue
337                provider = cast("MusicProvider", provider)
338                await provider.on_item_updated(library_item)
339        return library_item
340
341    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
342        """Delete library record from the database."""
343        db_id = int(item_id)  # ensure integer
344        library_item = await self.get_library_item(db_id)
345        assert library_item, f"Item does not exist: {db_id}"
346        # delete item
347        await self.mass.music.database.delete(
348            self.db_table,
349            {"item_id": db_id},
350        )
351        # update provider_mappings table
352        await self.mass.music.database.delete(
353            DB_TABLE_PROVIDER_MAPPINGS,
354            {"media_type": self.media_type.value, "item_id": db_id},
355        )
356        # cleanup external_id_lookup table
357        await self.mass.music.database.delete(
358            DB_TABLE_EXTERNAL_ID_LOOKUP,
359            {"media_type": self.media_type.value, "item_id": db_id},
360        )
361        # cleanup playlog table
362        await self.mass.music.database.delete(
363            DB_TABLE_PLAYLOG,
364            {
365                "media_type": self.media_type.value,
366                "item_id": db_id,
367                "provider": "library",
368            },
369        )
370        for prov_mapping in library_item.provider_mappings:
371            await self.mass.music.database.delete(
372                DB_TABLE_PLAYLOG,
373                {
374                    "media_type": self.media_type.value,
375                    "item_id": prov_mapping.item_id,
376                    "provider": prov_mapping.provider_instance,
377                },
378            )
379            # cleanup audio analysis rows for this provider mapping
380            for prov_key in (prov_mapping.provider_domain, prov_mapping.provider_instance):
381                await self.mass.music.database.delete(
382                    DB_TABLE_AUDIO_ANALYSIS,
383                    {
384                        "media_type": self.media_type.value,
385                        "item_id": prov_mapping.item_id,
386                        "provider": prov_key,
387                    },
388                )
389        # delete genre exclusions for this media item
390        await self.mass.music.database.delete(
391            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
392            {"media_type": self.media_type.value, "media_id": db_id},
393        )
394        # NOTE: this does not delete any references to this item in other records,
395        # this is handled/overridden in the mediatype specific controllers
396        # drop cached artwork for the removed item
397        for img in library_item.metadata.images or []:
398            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
399        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
400            self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, library_item.uri, library_item)
401        self.logger.debug("deleted item with id %s from database", db_id)
402
403    async def library_count(self, favorite_only: bool = False) -> int:
404        """
405        Return the number of items in the library.
406
407        Restricted to the providers the current user is allowed to see when that user
408        has a provider filter set.
409
410        :param favorite_only: Only count items marked as favorite.
411        """
412        query_parts: list[str] = []
413        query_params: dict[str, Any] = {}
414        if favorite_only:
415            query_parts.append("favorite = 1")
416        if provider_filter := self._ensure_provider_filter(None):
417            query_parts.append(
418                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
419            )
420        if not query_parts:
421            return await self.mass.music.database.get_count(self.db_table)
422        sql_query = f"SELECT item_id FROM {self.db_table} WHERE {' AND '.join(query_parts)}"
423        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
424
425    if TYPE_CHECKING:
426
427        @overload
428        async def library_items(
429            self,
430            favorite: bool | None = None,
431            search: str | None = None,
432            limit: int = 500,
433            offset: int = 0,
434            order_by: str = "sort_name",
435            provider: str | list[str] | None = None,
436            genre: int | list[int] | None = None,
437            played_only: bool = False,
438            *,
439            summary: bool = True,
440            collapse_collections: Literal[False] = False,
441            reachable_via: list[str] | None = None,
442            **kwargs: Any,
443        ) -> list[ItemCls]: ...
444
445        @overload
446        async def library_items(
447            self,
448            favorite: bool | None = None,
449            search: str | None = None,
450            limit: int = 500,
451            offset: int = 0,
452            order_by: str = "sort_name",
453            provider: str | list[str] | None = None,
454            genre: int | list[int] | None = None,
455            played_only: bool = False,
456            *,
457            summary: bool = True,
458            collapse_collections: Literal[True],
459            reachable_via: list[str] | None = None,
460            **kwargs: Any,
461        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
462
463        @overload
464        async def library_items(
465            self,
466            favorite: bool | None = None,
467            search: str | None = None,
468            limit: int = 500,
469            offset: int = 0,
470            order_by: str = "sort_name",
471            provider: str | list[str] | None = None,
472            genre: int | list[int] | None = None,
473            played_only: bool = False,
474            *,
475            summary: bool = True,
476            collapse_collections: bool,
477            reachable_via: list[str] | None = None,
478            **kwargs: Any,
479        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
480
481    async def library_items(  # noqa: PLR0913
482        self,
483        favorite: bool | None = None,
484        search: str | None = None,
485        limit: int = 500,
486        offset: int = 0,
487        order_by: str = "sort_name",
488        provider: str | list[str] | None = None,
489        genre: int | list[int] | None = None,
490        played_only: bool = False,
491        *,
492        summary: bool = True,
493        collapse_collections: bool = False,
494        reachable_via: list[str] | None = None,
495        **kwargs: Any,
496    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
497        """
498        Get the library items for this mediatype.
499
500        :param favorite: Filter by favorite status.
501        :param search: Filter by search query.
502        :param limit: Maximum number of items to return.
503        :param offset: Number of items to skip.
504        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
505        :param provider: Filter by provider instance ID (single string or list).
506        :param genre: Filter by genre id(s).
507        :param played_only: Only include items that have been played (last_played > 0).
508        :param summary: When True (default), return slim summary items containing only the
509            fields needed for a list view. Set to False to get fully hydrated items.
510        :param collapse_collections: Collapse available collections. Items in a collection won't
511            be returned individually.
512        :param reachable_via: Restrict results to items with a provider mapping reachable
513            through one of these provider instance ids (OR semantics), regardless of
514            whether that mapping is itself in that provider's own library. This is
515            independent of `provider`, which instead requires the *matched* mapping to
516            be in-library. None applies no filter; an explicit empty list, or a list
517            with no currently loaded/allowed instance, returns no items.
518        """
519        reachable_via = self._resolve_reachable_via(reachable_via)
520        if reachable_via is not None and not reachable_via:
521            return []
522        items = await self.get_library_items_by_query(
523            favorite=favorite,
524            search=search,
525            limit=limit,
526            offset=offset,
527            order_by=order_by,
528            provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
529            genre_ids=genre,
530            played_only=played_only,
531            in_library_only=True,
532            summary=summary,
533            collapse_collections=collapse_collections,
534            reachable_via=reachable_via,
535        )
536        if (
537            kwargs.get("_localized_fallback", True)
538            and search
539            and not items
540            and self.media_type in (MediaType.GENRE, MediaType.PLAYLIST)
541        ):
542            return await self._localized_search_fallback(
543                search,
544                limit=limit,
545                offset=offset,
546                favorite=favorite,
547                order_by=order_by,
548                provider=provider,
549                genre=genre,
550                summary=summary,
551                reachable_via=reachable_via,
552            )
553        return items
554
555    async def iter_library_items(
556        self,
557        favorite: bool | None = None,
558        search: str | None = None,
559        order_by: str = "sort_name",
560        provider: str | list[str] | None = None,
561        genre: int | list[int] | None = None,
562        library_items_only: bool = True,
563    ) -> AsyncGenerator[ItemCls]:
564        """Iterate all in-database items."""
565        limit: int = 500
566        offset: int = 0
567        if provider is not None:
568            provider_filter = provider if isinstance(provider, list) else [provider]
569        else:
570            provider_filter = None
571        while True:
572            next_items = await self.get_library_items_by_query(
573                favorite=favorite,
574                search=search,
575                genre_ids=genre,
576                limit=limit,
577                offset=offset,
578                order_by=order_by,
579                provider_filter=provider_filter,
580                in_library_only=library_items_only,
581            )
582            for item in next_items:
583                yield item
584            if len(next_items) < limit:
585                break
586            offset += limit
587
588    async def get(
589        self,
590        item_id: str,
591        provider_instance_id_or_domain: str,
592        allow_update_metadata: bool = True,
593    ) -> ItemCls:
594        """
595        Return (full) details for a single media item.
596
597        Tries to find the item in the library first, falling back to
598        fetching directly from the provider if not found.
599
600        :param item_id: The provider item id to fetch.
601        :param provider_instance_id_or_domain: The provider instance id or
602            domain to fetch the item from.
603        :param allow_update_metadata: Schedule a metadata refresh on access.
604            Set to False when fetching items in bulk (e.g. provider sync).
605        """
606        # always prefer the full library item if we have it
607        if library_item := await self.get_library_item_by_prov_id(
608            item_id,
609            provider_instance_id_or_domain,
610        ):
611            # schedule a refresh of the metadata on access of the item
612            # e.g. the item is being played or opened in the UI
613            if allow_update_metadata:
614                assert library_item.uri is not None
615                self.mass.metadata.schedule_update_metadata(library_item)
616            return library_item
617        # grab full details from the provider
618        return await self.get_provider_item(
619            item_id,
620            provider_instance_id_or_domain,
621        )
622
623    async def search(
624        self,
625        search_query: str,
626        provider_instance_id_or_domain: str,
627        limit: int = 25,
628    ) -> list[ItemCls]:
629        """Search database or provider with given query."""
630        # create safe search string
631        search_query = search_query.replace("/", " ").replace("'", "")
632        if provider_instance_id_or_domain == "library":
633            return await self.library_items(
634                search=search_query, limit=limit, summary=False, collapse_collections=False
635            )
636        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
637            return []
638        if prov.type != ProviderType.MUSIC:
639            return []
640        prov = cast("MusicProvider", prov)
641        if ProviderFeature.SEARCH not in prov.supported_features:
642            return []
643        if self.media_type not in prov.supported_media_types:
644            return []
645        searchresult = await prov.search(
646            search_query,
647            [self.media_type],
648            limit,
649        )
650        match self.media_type:
651            case MediaType.ARTIST:
652                return cast("list[ItemCls]", searchresult.artists)
653            case MediaType.ALBUM:
654                return cast("list[ItemCls]", searchresult.albums)
655            case MediaType.TRACK:
656                return cast("list[ItemCls]", searchresult.tracks)
657            case MediaType.PLAYLIST:
658                return cast("list[ItemCls]", searchresult.playlists)
659            case MediaType.AUDIOBOOK:
660                return cast("list[ItemCls]", searchresult.audiobooks)
661            case MediaType.PODCAST:
662                return cast("list[ItemCls]", searchresult.podcasts)
663            case MediaType.RADIO:
664                return cast("list[ItemCls]", searchresult.radio)
665            case _:
666                return []
667
668    async def get_collection(self, item_id: str) -> MediaCollection[ItemCls]:
669        """Get a single collection."""
670        name = get_collection_name_from_item_id(item_id)
671        query_params: dict[str, Any] = {"collection_name": name}
672        sql_query, base_query_params = self._build_final_query([], [], None, summary=False)
673        for key, value in base_query_params.items():
674            query_params.setdefault(key, value)
675        sql_query = await self._adapt_query_for_collections(
676            sql_query, query_params, summary=False, order_by=None, collection_name=name
677        )
678        db_rows = await self.mass.music.database.get_rows_from_query(
679            sql_query, query_params, limit=1, offset=0
680        )
681        if len(db_rows) != 1:
682            raise MediaNotFoundError(f"Collection {name} not found.")
683
684        return cast(
685            "MediaCollection[ItemCls]",
686            MediaCollection(
687                item_id=get_collection_item_id(db_rows[0]["name"], item_media_type=self.media_type),
688                name=db_rows[0]["name"],
689                provider="library",
690                provider_mappings=set(),
691                items=UniqueList(
692                    [
693                        self.item_cls.from_dict(self._parse_db_row(json_loads(x)))
694                        for x in json_loads(db_rows[0]["media_data"])
695                    ]
696                ),
697            ),
698        )
699
700    async def get_library_item(self, item_id: int | str) -> ItemCls:
701        """Get single library item by id."""
702        db_id = int(item_id)  # ensure integer
703        extra_query = f"WHERE {self.db_table}.item_id = :item_id"
704        for db_item in await self.get_library_items_by_query(
705            extra_query_parts=[extra_query],
706            extra_query_params={"item_id": db_id},
707            in_library_only=False,
708        ):
709            return db_item
710        msg = f"{self.media_type.value} not found in library: {db_id}"
711        raise MediaNotFoundError(msg)
712
713    async def get_library_item_by_prov_id(
714        self,
715        item_id: str,
716        provider_instance_id_or_domain: str,
717    ) -> ItemCls | None:
718        """Get the library item for the given provider item, if present."""
719        assert item_id
720        assert provider_instance_id_or_domain
721        if provider_instance_id_or_domain == "library":
722            try:
723                return await self.get_library_item(item_id)
724            except MediaNotFoundError:
725                return None
726        for item in await self.get_library_items_by_prov_id(
727            provider_instance_id_or_domain=provider_instance_id_or_domain,
728            provider_item_id=item_id,
729        ):
730            return item
731        return None
732
733    @final
734    async def get_library_item_by_prov_mappings(
735        self,
736        provider_mappings: Iterable[ProviderMapping],
737    ) -> ItemCls | None:
738        """Get the library item for the given provider_instance."""
739        # always prefer provider instance first
740        for mapping in provider_mappings:
741            for item in await self.get_library_items_by_prov_id(
742                provider_instance=mapping.provider_instance,
743                provider_item_id=mapping.item_id,
744            ):
745                return item
746        # check by domain too
747        for mapping in provider_mappings:
748            for item in await self.get_library_items_by_prov_id(
749                provider_domain=mapping.provider_domain,
750                provider_item_id=mapping.item_id,
751            ):
752                return item
753        return None
754
755    @final
756    async def get_library_item_sync_details(
757        self,
758        provider_mappings: Iterable[ProviderMapping],
759    ) -> LibraryItemSyncDetails | None:
760        """
761        Get a lightweight sync snapshot of the library item for the given provider mappings.
762
763        Returns only the scalar columns and raw provider mapping rows the library sync
764        needs for its change detection, without hydrating a full MediaItem object.
765        Resolution order matches get_library_item_by_prov_mappings (instance first,
766        then domain).
767        """
768        extra_columns, extra_joins, extra_params = self._sync_details_query_parts()
769        base_sql = f"""
770            SELECT
771                {self.db_table}.item_id,
772                {self.db_table}.favorite,
773                {self.db_table}.timestamp_added,
774                (SELECT JSON_GROUP_ARRAY(
775                    json_object(
776                        'item_id', pm.provider_item_id,
777                        'provider_domain', pm.provider_domain,
778                        'provider_instance', pm.provider_instance,
779                        'available', pm.available,
780                        'in_library', pm.in_library,
781                        'is_unique', pm.is_unique
782                    )) FROM provider_mappings pm WHERE pm.item_id = {self.db_table}.item_id
783                        AND pm.media_type = '{self.media_type.value}') AS provider_mappings
784                {extra_columns}
785            FROM {self.db_table}
786            {extra_joins}
787            WHERE {self.db_table}.item_id IN (
788                SELECT item_id FROM provider_mappings
789                WHERE provider_mappings.media_type = '{self.media_type.value}'
790                AND provider_mappings.{{prov_column}} = :prov_id
791                AND provider_mappings.provider_item_id = :prov_item_id
792            )
793        """
794        # always prefer provider instance first, then domain
795        # (same resolution order as get_library_item_by_prov_mappings)
796        for prov_column in ("provider_instance", "provider_domain"):
797            for mapping in provider_mappings:
798                for db_row in await self.mass.music.database.get_rows_from_query(
799                    base_sql.format(prov_column=prov_column),
800                    {
801                        **extra_params,
802                        "prov_id": getattr(mapping, prov_column),
803                        "prov_item_id": mapping.item_id,
804                    },
805                    limit=1,
806                ):
807                    return self._parse_sync_details_row(db_row)
808        return None
809
810    @final
811    async def get_library_items_by_external_id(
812        self,
813        external_id: str,
814        external_id_type: ExternalID | None = None,
815        *,
816        limit: int | None,
817    ) -> list[ItemCls]:
818        """
819        Get library items for the given external identifier.
820
821        :param external_id: External identifier value to look up.
822        :param external_id_type: Optional identifier type.
823        :param limit: Maximum number of library items to return, or None for all matches.
824        """
825        if external_id_type:
826            lookup_values = external_id_lookup_values(external_id_type, external_id)
827        else:
828            lookup_values = external_id_lookup_values_untyped(external_id)
829        subquery_parts = [
830            "media_type = :ext_id_media_type",
831            "external_id IN :external_ids",
832        ]
833        query_params: dict[str, Any] = {
834            "ext_id_media_type": self.media_type.value,
835            "external_ids": lookup_values,
836        }
837        if external_id_type:
838            subquery_parts.append("external_id_type = :external_id_type")
839            query_params["external_id_type"] = str(external_id_type)
840        subquery = (
841            f"SELECT item_id FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
842            f"WHERE {' AND '.join(subquery_parts)}"
843        )
844        query = f"{self.db_table}.item_id IN ({subquery})"
845        if limit is not None:
846            limited_items = await self.get_library_items_by_query(
847                limit=limit,
848                extra_query_parts=[query],
849                extra_query_params=query_params,
850            )
851            return sorted(limited_items, key=lambda item: int(item.item_id))
852
853        all_items: list[ItemCls] = []
854        offset = 0
855        page_size = 500
856        while page := await self.get_library_items_by_query(
857            limit=page_size,
858            offset=offset,
859            extra_query_parts=[query],
860            extra_query_params=query_params,
861        ):
862            all_items.extend(page)
863            if len(page) < page_size:
864                break
865            offset += page_size
866        return sorted(all_items, key=lambda item: int(item.item_id))
867
868    @final
869    async def get_library_item_by_external_id(
870        self, external_id: str, external_id_type: ExternalID | None = None
871    ) -> ItemCls | None:
872        """Get the first library item for the given external id, if present."""
873        items = await self.get_library_items_by_external_id(external_id, external_id_type, limit=1)
874        return items[0] if items else None
875
876    @final
877    async def get_library_items_by_external_ids(
878        self, external_ids: set[tuple[ExternalID, str]]
879    ) -> list[ItemCls]:
880        """Get all library items matching any of the given external identifiers."""
881        result: dict[str, ItemCls] = {}
882        for external_id_type, external_id in sorted(external_ids, key=external_id_sort_key):
883            for item in await self.get_library_items_by_external_id(
884                external_id, external_id_type, limit=None
885            ):
886                result.setdefault(item.item_id, item)
887        return list(result.values())
888
889    @final
890    async def get_library_item_by_external_ids(
891        self, external_ids: set[tuple[ExternalID, str]]
892    ) -> ItemCls | None:
893        """Get the library item for (one of) the given external ids."""
894        items = await self.get_library_items_by_external_ids(external_ids)
895        return items[0] if items else None
896
897    @final
898    async def get_library_items_by_prov_id(
899        self,
900        provider_domain: str | None = None,
901        provider_instance: str | None = None,
902        provider_instance_id_or_domain: str | None = None,
903        provider_item_id: str | None = None,
904        provider_item_ids: list[str] | None = None,
905        limit: int = 500,
906        offset: int = 0,
907    ) -> list[ItemCls]:
908        """
909        Fetch all records from library for given provider.
910
911        :param provider_item_ids: When given, batch-match this list of provider
912            item ids in a single query (the plural form of provider_item_id);
913            takes precedence over provider_item_id when both are passed. An
914            empty list matches nothing (distinct from None, which applies no
915            item-id filter).
916        """
917        assert provider_instance_id_or_domain != "library"
918        assert provider_domain != "library"
919        assert provider_instance != "library"
920        if provider_item_ids is not None and not provider_item_ids:
921            return []
922        subquery_parts: list[str] = []
923        query_params: dict[str, Any] = {}
924        if provider_instance:
925            query_params = {"prov_id": provider_instance}
926            subquery_parts.append("provider_mappings.provider_instance = :prov_id")
927        elif provider_domain:
928            query_params = {"prov_id": provider_domain}
929            subquery_parts.append("provider_mappings.provider_domain = :prov_id")
930        else:
931            query_params = {"prov_id": provider_instance_id_or_domain}
932            subquery_parts.append(
933                "(provider_mappings.provider_instance = :prov_id "
934                "OR provider_mappings.provider_domain = :prov_id)"
935            )
936        if provider_item_ids:
937            placeholders = ", ".join(f":item_id_{i}" for i in range(len(provider_item_ids)))
938            subquery_parts.append(f"provider_mappings.provider_item_id IN ({placeholders})")
939            for i, item_id in enumerate(provider_item_ids):
940                query_params[f"item_id_{i}"] = item_id
941        elif provider_item_id:
942            subquery_parts.append("provider_mappings.provider_item_id = :item_id")
943            query_params["item_id"] = provider_item_id
944        subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}"
945        query = f"WHERE {self.db_table}.item_id IN ({subquery})"
946        return await self.get_library_items_by_query(
947            limit=limit,
948            offset=offset,
949            extra_query_parts=[query],
950            extra_query_params=query_params,
951            in_library_only=False,
952        )
953
954    @final
955    async def iter_library_items_by_prov_id(
956        self,
957        provider_instance_id_or_domain: str,
958        provider_item_id: str | None = None,
959    ) -> AsyncGenerator[ItemCls]:
960        """Iterate all records from database for given provider."""
961        limit: int = 500
962        offset: int = 0
963        while True:
964            next_items = await self.get_library_items_by_prov_id(
965                provider_instance_id_or_domain=provider_instance_id_or_domain,
966                provider_item_id=provider_item_id,
967                limit=limit,
968                offset=offset,
969            )
970            for item in next_items:
971                yield item
972            if len(next_items) < limit:
973                break
974            offset += limit
975
976    @final
977    async def set_favorite(self, item_id: str | int, favorite: bool) -> None:
978        """Set the favorite bool on a database item."""
979        db_id = int(item_id)  # ensure integer
980        library_item = await self.get_library_item(db_id)
981        if library_item.favorite == favorite:
982            return
983        match = {"item_id": db_id}
984        await self.mass.music.database.update(self.db_table, match, {"favorite": favorite})
985        library_item = await self.get_library_item(db_id)
986        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
987
988    @guard_single_request
989    @final
990    async def get_provider_item(
991        self,
992        item_id: str,
993        provider_instance_id_or_domain: str,
994        force_refresh: bool = False,
995        fallback: ItemMapping | ItemCls | None = None,
996    ) -> ItemCls:
997        """Return item details for the given provider item id."""
998        if provider_instance_id_or_domain == "library":
999            return await self.get_library_item(item_id)
1000        if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
1001            raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
1002        if provider := self.mass.get_provider(provider_instance_id_or_domain):
1003            provider = cast("MusicProvider | PluginProvider", provider)
1004            with suppress(MediaNotFoundError):
1005                async with self.mass.cache.handle_refresh(force_refresh):
1006                    if self.media_type == MediaType.PLAYLIST:
1007                        return cast("ItemCls", await provider.get_playlist(item_id))
1008                    music_prov = cast("MusicProvider", provider)
1009                    if self.media_type == MediaType.ARTIST:
1010                        return cast("ItemCls", await music_prov.get_artist(item_id))
1011                    if self.media_type == MediaType.ALBUM:
1012                        return cast("ItemCls", await music_prov.get_album(item_id))
1013                    if self.media_type == MediaType.TRACK:
1014                        return cast("ItemCls", await music_prov.get_track(item_id))
1015                    if self.media_type == MediaType.RADIO:
1016                        return cast("ItemCls", await music_prov.get_radio(item_id))
1017                    if self.media_type == MediaType.AUDIOBOOK:
1018                        return cast("ItemCls", await music_prov.get_audiobook(item_id))
1019                    if self.media_type == MediaType.PODCAST:
1020                        return cast("ItemCls", await music_prov.get_podcast(item_id))
1021        # if we reach this point all possibilities failed and the item could not be found.
1022        # There is a possibility that the (streaming) provider changed the id of the item
1023        # so we return the previous details (if we have any) marked as unavailable, so
1024        # at least we have the possibility to sort out the new id through matching logic.
1025        fallback = fallback or await self.get_library_item_by_prov_id(
1026            item_id, provider_instance_id_or_domain
1027        )
1028        if (
1029            fallback
1030            and isinstance(fallback, ItemMapping)
1031            and (fallback_provider := self.mass.get_provider(fallback.provider))
1032        ):
1033            # fallback is a ItemMapping, try to convert to full item
1034            with suppress(LookupError, TypeError, ValueError):
1035                return cast(
1036                    "ItemCls",
1037                    self.item_cls.from_dict(
1038                        {
1039                            **fallback.to_dict(),
1040                            "provider_mappings": [
1041                                {
1042                                    "item_id": fallback.item_id,
1043                                    "provider_domain": fallback_provider.domain,
1044                                    "provider_instance": fallback_provider.instance_id,
1045                                    "available": fallback.available,
1046                                }
1047                            ],
1048                        }
1049                    ),
1050                )
1051        if fallback:
1052            # simply return the fallback item
1053            return cast("ItemCls", fallback)
1054        # all options exhausted, we really can not find this item
1055        msg = (
1056            f"{self.media_type.value}://{item_id} not "
1057            f"found on provider {provider_instance_id_or_domain}"
1058        )
1059        raise MediaNotFoundError(msg)
1060
1061    @final
1062    async def add_provider_mapping(
1063        self, item_id: str | int, provider_mapping: ProviderMapping
1064    ) -> None:
1065        """Add provider mapping to existing library item."""
1066        await self.add_provider_mappings(item_id, [provider_mapping])
1067
1068    @final
1069    async def merge_library_items(
1070        self, target_item_id: str | int, source_item_id: str | int
1071    ) -> ItemCls:
1072        """
1073        Merge one library item into another and return the target item.
1074
1075        The explicit target is the deterministic winner. Its current values stay authoritative
1076        where the normal non-overwrite model update keeps them; the source is merged as the
1077        incoming update. All source state is transferred before the source row is deleted.
1078
1079        :param target_item_id: Library ID of the item that remains after the merge.
1080        :param source_item_id: Library ID of the duplicate item that is removed after transfer.
1081        :raises InvalidDataError: When the IDs are identical or do not belong to this media type.
1082        """
1083        target_id = int(target_item_id)
1084        source_id = int(source_item_id)
1085        if target_id == source_id:
1086            msg = "Cannot merge a library item into itself"
1087            raise InvalidDataError(msg)
1088        async with self._db_add_lock:
1089            return await self._merge_library_items_batched(target_id, source_id)
1090
1091    @final
1092    async def add_provider_mappings(
1093        self, item_id: str | int, provider_mappings: Iterable[ProviderMapping]
1094    ) -> None:
1095        """
1096        Add provider mappings to existing library item.
1097
1098        :param item_id: The library item ID to add mappings to.
1099        :param provider_mappings: The provider mappings to add.
1100        """
1101        db_id = int(item_id)  # ensure integer
1102        mappings = set(provider_mappings)
1103        if not mappings:
1104            return
1105        async with self._db_add_lock:
1106            library_item = await self.get_library_item(db_id)
1107            while True:
1108                conflicting_item = None
1109                for mapping in mappings:
1110                    existing_item = await self.get_library_item_by_prov_id(
1111                        mapping.item_id, mapping.provider_instance
1112                    )
1113                    if existing_item and int(existing_item.item_id) != db_id:
1114                        conflicting_item = existing_item
1115                        break
1116                if conflicting_item is None:
1117                    break
1118                self.logger.debug(
1119                    "merging item id %s into item id %s based on provider mapping",
1120                    conflicting_item.item_id,
1121                    library_item.item_id,
1122                )
1123                library_item = await self._merge_library_items_batched(
1124                    db_id, int(conflicting_item.item_id)
1125                )
1126
1127            new_mappings = mappings.difference(library_item.provider_mappings)
1128            if not new_mappings:
1129                return
1130            library_item.provider_mappings.update(new_mappings)
1131            self.mass.music.match_provider_instances(library_item)
1132            await self.set_provider_mappings(db_id, library_item.provider_mappings)
1133            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
1134
1135    @final
1136    async def update_provider_mapping(
1137        self,
1138        item_id: str | int,
1139        provider_instance_id: str,
1140        provider_item_id: str,
1141        *,
1142        available: bool | Any = UNSET,
1143        in_library: bool | Any = UNSET,
1144        is_unique: bool | None | Any = UNSET,
1145        url: str | None | Any = UNSET,
1146        details: str | None | Any = UNSET,
1147        audio_format: AudioFormat | Any = UNSET,
1148    ) -> None:
1149        """Update an existing provider mapping for a library item."""
1150        db_id = int(item_id)  # ensure integer
1151        library_item = await self.get_library_item(db_id)
1152
1153        # find the current mapping (strictly by provider instance + provider item id)
1154        cur_mapping: ProviderMapping | None = None
1155        for mapping in library_item.provider_mappings:
1156            if (
1157                mapping.provider_instance == provider_instance_id
1158                and mapping.item_id == provider_item_id
1159            ):
1160                cur_mapping = mapping
1161                break
1162        if cur_mapping is None:
1163            msg = (
1164                f"Provider mapping {provider_instance_id}/{provider_item_id} "
1165                f"not found for item {db_id}"
1166            )
1167            raise MediaNotFoundError(msg)
1168
1169        # guard against nulls for NOT NULL columns
1170        if available is None:
1171            available = UNSET
1172        if in_library is None:
1173            in_library = UNSET
1174
1175        updates: dict[str, Any] = {}
1176        if available is not UNSET:
1177            updates["available"] = bool(available)
1178        if in_library is not UNSET:
1179            updates["in_library"] = bool(in_library)
1180        if is_unique is not UNSET:
1181            updates["is_unique"] = is_unique
1182        if url is not UNSET:
1183            updates["url"] = url
1184        if details is not UNSET:
1185            updates["details"] = details
1186        if audio_format is not UNSET:
1187            updates["audio_format"] = serialize_to_json(audio_format)
1188
1189        if not updates:
1190            return
1191
1192        match = {
1193            "media_type": self.media_type.value,
1194            "item_id": db_id,
1195            "provider_instance": provider_instance_id,
1196            "provider_item_id": provider_item_id,
1197        }
1198        await self.mass.music.database.update(DB_TABLE_PROVIDER_MAPPINGS, match, updates)
1199
1200        # Re-fetch the updated item so the event payload reflects persisted DB state.
1201        updated_item = await self.get_library_item(db_id)
1202        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, updated_item.uri, updated_item)
1203
1204    @final
1205    async def remove_provider_mapping(
1206        self, item_id: str | int, provider_instance_id: str, provider_item_id: str
1207    ) -> None:
1208        """Remove provider mapping(s) from item."""
1209        db_id = int(item_id)  # ensure integer
1210        try:
1211            library_item = await self.get_library_item(db_id)
1212        except MediaNotFoundError:
1213            # edge case: already deleted / race condition
1214            return
1215
1216        remaining_mappings = {
1217            x
1218            for x in library_item.provider_mappings
1219            if not (x.provider_instance == provider_instance_id and x.item_id == provider_item_id)
1220        }
1221        if not remaining_mappings:
1222            # this was the last mapping, so remove the entire library item, which also
1223            # clears its provider mapping rows. Dropping those rows up front would leave
1224            # the item behind without any mappings if the removal itself fails.
1225            with suppress(MediaNotFoundError):
1226                await self.remove_item_from_library(db_id)
1227            return
1228
1229        # update provider_mappings table
1230        await self.mass.music.database.delete(
1231            DB_TABLE_PROVIDER_MAPPINGS,
1232            {
1233                "media_type": self.media_type.value,
1234                "item_id": db_id,
1235                "provider_instance": provider_instance_id,
1236                "provider_item_id": provider_item_id,
1237            },
1238        )
1239        # cleanup playlog table
1240        await self.mass.music.database.delete(
1241            DB_TABLE_PLAYLOG,
1242            {
1243                "media_type": self.media_type.value,
1244                "item_id": provider_item_id,
1245                "provider": provider_instance_id,
1246            },
1247        )
1248        library_item.provider_mappings = remaining_mappings
1249        # if this was the last mapping for the provider instance, strip any artwork
1250        # that belonged to it (e.g. local file paths that are no longer resolvable)
1251        images_changed = not any(
1252            x.provider_instance == provider_instance_id for x in remaining_mappings
1253        ) and await self._remove_provider_images(db_id, provider_instance_id)
1254        self.logger.debug(
1255            "removed provider_mapping %s/%s from item id %s",
1256            provider_instance_id,
1257            provider_item_id,
1258            db_id,
1259        )
1260        # the removed provider mapping is itself a change to the item, so always notify
1261        # (unless suppressed during a bulk cleanup); re-fetch first when images were
1262        # stripped so the event payload stays accurate
1263        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1264            event_item = await self.get_library_item(db_id) if images_changed else library_item
1265            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1266
1267    @final
1268    async def remove_provider_mappings(self, item_id: str | int, provider_instance_id: str) -> None:
1269        """Remove all provider mappings from an item."""
1270        db_id = int(item_id)  # ensure integer
1271        try:
1272            library_item = await self.get_library_item(db_id)
1273        except MediaNotFoundError:
1274            # edge case: already deleted / race condition, just drop any leftover rows
1275            await self.mass.music.database.delete(
1276                DB_TABLE_PROVIDER_MAPPINGS,
1277                {
1278                    "media_type": self.media_type.value,
1279                    "item_id": db_id,
1280                    "provider_instance": provider_instance_id,
1281                },
1282            )
1283            return
1284
1285        remaining_mappings = {
1286            x for x in library_item.provider_mappings if x.provider_instance != provider_instance_id
1287        }
1288        if not remaining_mappings:
1289            # these were the last mappings, so remove the entire library item, which also
1290            # clears its provider mapping rows. Dropping those rows up front would leave
1291            # the item behind without any mappings if the removal itself fails.
1292            with suppress(MediaNotFoundError):
1293                await self.remove_item_from_library(db_id)
1294            return
1295
1296        # update provider_mappings table
1297        await self.mass.music.database.delete(
1298            DB_TABLE_PROVIDER_MAPPINGS,
1299            {
1300                "media_type": self.media_type.value,
1301                "item_id": db_id,
1302                "provider_instance": provider_instance_id,
1303            },
1304        )
1305        library_item.provider_mappings = remaining_mappings
1306        # the item is kept (it still has other providers), but it may carry artwork
1307        # that belonged to the removed provider (e.g. local file paths that are no
1308        # longer resolvable), so strip those images from the stored metadata
1309        images_changed = await self._remove_provider_images(db_id, provider_instance_id)
1310        self.logger.debug(
1311            "removed all provider mappings for provider %s from item id %s",
1312            provider_instance_id,
1313            db_id,
1314        )
1315        # the removed provider mapping(s) are themselves a change to the item, so
1316        # always notify (unless suppressed during a bulk cleanup); re-fetch first when
1317        # images were stripped so the event payload stays accurate
1318        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1319            event_item = await self.get_library_item(db_id) if images_changed else library_item
1320            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1321
1322    @final
1323    async def set_provider_mappings(
1324        self,
1325        item_id: str | int,
1326        provider_mappings: Iterable[ProviderMapping],
1327        overwrite: bool = False,
1328    ) -> None:
1329        """
1330        Update the provider_mappings table for the media item.
1331
1332        An empty set of mappings never clears the stored rows: an item without any
1333        mapping can not be played or resolved.
1334        """
1335        db_id = int(item_id)  # ensure integer
1336        prov_map_objs: list[dict[str, Any]] = []
1337        for provider_mapping in provider_mappings:
1338            prov_map_obj = {
1339                "media_type": self.media_type.value,
1340                "item_id": db_id,
1341                "provider_domain": provider_mapping.provider_domain,
1342                "provider_instance": provider_mapping.provider_instance,
1343                "provider_item_id": provider_mapping.item_id,
1344                "available": provider_mapping.available,
1345                "audio_format": serialize_to_json(provider_mapping.audio_format),
1346            }
1347            for key in ("url", "details", "in_library", "is_unique"):
1348                if (value := getattr(provider_mapping, key, None)) is not None:
1349                    prov_map_obj[key] = value
1350            prov_map_objs.append(prov_map_obj)
1351        if not prov_map_objs:
1352            if overwrite:
1353                # a caller asking to replace all mappings with none is a bug,
1354                # so keep the stored rows and make the attempt visible
1355                self.logger.warning(
1356                    "Ignoring request to clear all provider mappings of %s item id %s",
1357                    self.media_type.value,
1358                    db_id,
1359                )
1360            return
1361        if overwrite:
1362            # on overwrite, clear the provider_mappings table first
1363            # this is done for filesystem provider changing the path (and thus item_id)
1364            await self.mass.music.database.delete(
1365                DB_TABLE_PROVIDER_MAPPINGS,
1366                {"media_type": self.media_type.value, "item_id": db_id},
1367            )
1368        await self.mass.music.database.upsert_many(
1369            DB_TABLE_PROVIDER_MAPPINGS,
1370            prov_map_objs,
1371        )
1372
1373    @final
1374    async def set_external_ids(
1375        self,
1376        item_id: str | int,
1377        external_ids: Iterable[tuple[ExternalID, str]],
1378    ) -> None:
1379        """
1380        Update the external_id_lookup table rows for the media item.
1381
1382        An empty set never clears the stored rows: identifiers are the strongest
1383        evidence available when matching an item across providers.
1384        """
1385        db_id = int(item_id)  # ensure integer
1386        if not (external_ids := normalize_external_ids(external_ids)):
1387            return
1388        await self.mass.music.database.delete(
1389            DB_TABLE_EXTERNAL_ID_LOOKUP,
1390            {"media_type": self.media_type.value, "item_id": db_id},
1391        )
1392        await self.mass.music.database.upsert_many(
1393            DB_TABLE_EXTERNAL_ID_LOOKUP,
1394            [
1395                {
1396                    "media_type": self.media_type.value,
1397                    "external_id_type": external_id_type,
1398                    "external_id": external_id,
1399                    "item_id": db_id,
1400                }
1401                for external_id_type, external_id in external_ids
1402            ],
1403        )
1404
1405    @abstractmethod
1406    async def match_providers(self, db_item: ItemCls) -> None:
1407        """
1408        Try to find match on all (streaming) providers for the provided (database) item.
1409
1410        This is used to link objects of different providers/qualities together.
1411        """
1412
1413    if TYPE_CHECKING:
1414
1415        @overload
1416        async def get_library_items_by_query(
1417            self,
1418            favorite: bool | None = None,
1419            search: str | None = None,
1420            limit: int = 500,
1421            offset: int = 0,
1422            order_by: str | None = None,
1423            provider_filter: list[str] | None = None,
1424            extra_query_parts: list[str] | None = None,
1425            extra_query_params: dict[str, Any] | None = None,
1426            extra_join_parts: list[str] | None = None,
1427            genre_ids: int | list[int] | None = None,
1428            played_only: bool = False,
1429            in_library_only: bool = False,
1430            summary: bool = False,
1431            *,
1432            collapse_collections: Literal[True],
1433            reachable_via: list[str] | None = None,
1434        ) -> list[ItemCls | MediaCollection[ItemCls]]: ...
1435
1436        @overload
1437        async def get_library_items_by_query(
1438            self,
1439            favorite: bool | None = None,
1440            search: str | None = None,
1441            limit: int = 500,
1442            offset: int = 0,
1443            order_by: str | None = None,
1444            provider_filter: list[str] | None = None,
1445            extra_query_parts: list[str] | None = None,
1446            extra_query_params: dict[str, Any] | None = None,
1447            extra_join_parts: list[str] | None = None,
1448            genre_ids: int | list[int] | None = None,
1449            played_only: bool = False,
1450            in_library_only: bool = False,
1451            summary: bool = False,
1452            *,
1453            collapse_collections: Literal[False] = False,
1454            reachable_via: list[str] | None = None,
1455        ) -> list[ItemCls]: ...
1456
1457        @overload
1458        async def get_library_items_by_query(
1459            self,
1460            favorite: bool | None = None,
1461            search: str | None = None,
1462            limit: int = 500,
1463            offset: int = 0,
1464            order_by: str | None = None,
1465            provider_filter: list[str] | None = None,
1466            extra_query_parts: list[str] | None = None,
1467            extra_query_params: dict[str, Any] | None = None,
1468            extra_join_parts: list[str] | None = None,
1469            genre_ids: int | list[int] | None = None,
1470            played_only: bool = False,
1471            in_library_only: bool = False,
1472            summary: bool = False,
1473            *,
1474            collapse_collections: bool,
1475            reachable_via: list[str] | None = None,
1476        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
1477
1478    @final
1479    async def get_library_items_by_query(  # noqa: PLR0913
1480        self,
1481        favorite: bool | None = None,
1482        search: str | None = None,
1483        limit: int = 500,
1484        offset: int = 0,
1485        order_by: str | None = None,
1486        provider_filter: list[str] | None = None,
1487        extra_query_parts: list[str] | None = None,
1488        extra_query_params: dict[str, Any] | None = None,
1489        extra_join_parts: list[str] | None = None,
1490        genre_ids: int | list[int] | None = None,
1491        played_only: bool = False,
1492        in_library_only: bool = False,
1493        summary: bool = False,
1494        *,
1495        collapse_collections: bool = False,
1496        reachable_via: list[str] | None = None,
1497    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
1498        """Fetch MediaItem records from database by building the query."""
1499        query_params = dict(extra_query_params) if extra_query_params else {}
1500        query_parts: list[str] = list(extra_query_parts) if extra_query_parts else []
1501        join_parts: list[str] = list(extra_join_parts) if extra_join_parts else []
1502        search = self._preprocess_search(search)
1503        genre_ids = self._preprocess_genre_ids(genre_ids)
1504        # create special performant random query
1505        if order_by and order_by.startswith("random"):
1506            self._apply_random_subquery(
1507                query_parts=query_parts,
1508                query_params=query_params,
1509                join_parts=join_parts,
1510                favorite=favorite,
1511                search=search if not collapse_collections else None,
1512                genre_ids=genre_ids,
1513                provider_filter=provider_filter,
1514                played_only=played_only,
1515                limit=limit,
1516                in_library_only=in_library_only,
1517                reachable_via=reachable_via,
1518            )
1519        else:
1520            # apply filters
1521            self._apply_filters(
1522                query_parts=query_parts,
1523                query_params=query_params,
1524                favorite=favorite,
1525                search=search if not collapse_collections else None,
1526                genre_ids=genre_ids,
1527                provider_filter=provider_filter,
1528                played_only=played_only,
1529                in_library_only=in_library_only,
1530                reachable_via=reachable_via,
1531            )
1532        # build and execute final query
1533        sql_query, base_query_params = self._build_final_query(
1534            query_parts, join_parts, order_by, summary=summary
1535        )
1536        # base query params act as defaults: callers may override them via extra_query_params
1537        for key, value in base_query_params.items():
1538            query_params.setdefault(key, value)
1539
1540        if collapse_collections:
1541            if search:
1542                query_params["search"] = f"%{search}%"
1543            sql_query = await self._adapt_query_for_collections(
1544                sql_query, query_params, summary=summary, order_by=order_by, search=search
1545            )
1546
1547        db_rows = await self.mass.music.database.get_rows_from_query(
1548            sql_query, query_params, limit=limit, offset=offset
1549        )
1550        if collapse_collections:
1551            items: list[ItemCls | MediaCollection[ItemCls]] = []
1552
1553            def _parse_method(x: str) -> ItemCls:
1554                if summary:
1555                    return cast("ItemCls", self._parse_summary_row(json_loads(x)))
1556                return cast(
1557                    "ItemCls",
1558                    self.item_cls.from_dict(self._parse_db_row(json_loads(x))),
1559                )
1560
1561            for db_row in db_rows:
1562                if db_row["type"] == "single":
1563                    items.append(_parse_method(db_row["media_data"]))
1564                elif db_row["type"] == "collection":
1565                    items.append(
1566                        MediaCollection[ItemCls](
1567                            item_id=get_collection_item_id(
1568                                db_row["name"], item_media_type=self.media_type
1569                            ),
1570                            name=db_row["name"],
1571                            provider="library",
1572                            provider_mappings=set(),
1573                            items=UniqueList(
1574                                [_parse_method(x) for x in json_loads(db_row["media_data"])]
1575                            ),
1576                        )
1577                    )
1578            return items
1579        if summary:
1580            return [cast("ItemCls", self._parse_summary_row(db_row)) for db_row in db_rows]
1581        return [
1582            cast("ItemCls", self.item_cls.from_dict(self._parse_db_row(db_row)))
1583            for db_row in db_rows
1584        ]
1585
1586    @final
1587    async def _get_library_item_by_match(self, item: ItemCls | ItemMapping) -> int | None:
1588        if item.provider == "library":
1589            return int(item.item_id)
1590        # search by provider mappings if item is ItemMapping
1591        if isinstance(item, ItemMapping):
1592            if cur_item := await self.get_library_item_by_prov_id(item.item_id, item.provider):
1593                return int(cur_item.item_id)
1594
1595        # for all other items that are MediaItemType, check provider_mappings if it exists
1596        provider_mappings = getattr(item, "provider_mappings", None)
1597        if provider_mappings:
1598            if cur_item := await self.get_library_item_by_prov_mappings(provider_mappings):
1599                return int(cur_item.item_id)
1600        # fetch candidates per external id (best identifier first) and stop at the
1601        # first verified match; external identifiers may be reused, so verify
1602        # every candidate before accepting it
1603        seen_item_ids: set[str] = set()
1604        for external_id_type, external_id in sorted(item.external_ids, key=external_id_sort_key):
1605            for cur_item in await self.get_library_items_by_external_id(
1606                external_id, external_id_type, limit=None
1607            ):
1608                if cur_item.item_id in seen_item_ids:
1609                    continue
1610                seen_item_ids.add(cur_item.item_id)
1611                if await self._confirm_library_candidate(cur_item, item):
1612                    return int(cur_item.item_id)
1613        # search by normalized exact name match
1614        query = (
1615            f"{self.db_table}.search_name IN :search_names "
1616            f"OR {self.db_table}.search_sort_name = :search_sort_name"
1617        )
1618        query_params = {
1619            "search_names": self._library_match_names(item),
1620            "search_sort_name": create_safe_string(item.sort_name or "", True, True),
1621        }
1622        for db_item in await self.get_library_items_by_query(
1623            extra_query_parts=[query], extra_query_params=query_params
1624        ):
1625            if await self._confirm_library_candidate(db_item, item):
1626                return int(db_item.item_id)
1627        return None
1628
1629    def _library_match_names(self, item: ItemCls | ItemMapping) -> list[str]:
1630        """
1631        Return the normalized names a library row for this item may be stored under.
1632
1633        Override in a subclass when a media type's title carries formatting that the
1634        stored name keeps but its identity comparison ignores.
1635        """
1636        return [create_safe_string(item.name, True, True)]
1637
1638    async def _confirm_library_candidate(
1639        self, db_item: ItemCls, item: ItemCls | ItemMapping
1640    ) -> bool:
1641        """
1642        Return True if a library candidate is the same item as the one being added.
1643
1644        Override in a subclass to confirm a candidate that the items' own metadata
1645        cannot decide on with additional evidence.
1646
1647        :param db_item: Existing library item that matched on an external id or name.
1648        :param item: The (provider) item that is being added to the library.
1649        """
1650        return bool(compare_media_item(db_item, item, True))
1651
1652    def _external_ids_query(
1653        self, media_type: MediaType | None = None, table_alias: str | None = None
1654    ) -> str:
1655        """
1656        Return a subquery that selects the external ids of a media item as a JSON array.
1657
1658        :param media_type: Media type to select the external ids for, defaults to
1659            this controller's media type.
1660        :param table_alias: (Aliased) table name the subquery correlates against,
1661            defaults to this controller's table.
1662        """
1663        media_type = media_type or self.media_type
1664        table_alias = table_alias or self.db_table
1665        return (
1666            f"(SELECT JSON_GROUP_ARRAY(json_array("
1667            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1668            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1669            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1670            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1671            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1672        )
1673
1674    def _provider_mappings_query(self) -> str:
1675        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1676        return f"""(SELECT JSON_GROUP_ARRAY(
1677            json_object(
1678                'item_id', pm.provider_item_id,
1679                'provider_domain', pm.provider_domain,
1680                'provider_instance', pm.provider_instance,
1681                'available', pm.available,
1682                'audio_format', json(pm.audio_format),
1683                'url', pm.url,
1684                'details', pm.details,
1685                'in_library', pm.in_library,
1686                'is_unique', pm.is_unique
1687            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1688            WHERE pm.item_id = {self.db_table}.item_id
1689            AND pm.media_type = '{self.media_type.value}')"""
1690
1691    def _artist_mappings_summary_query(
1692        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1693    ) -> str:
1694        """
1695        Return a subquery selecting the slim artist mappings JSON of a summary row.
1696
1697        :param m2m_table: The many-to-many table linking artists to this media type.
1698        :param m2m_key: The column in the m2m table referencing this media type's item id.
1699        :param include_artist_type: Also select the artist_type of each artist.
1700        """
1701        artist_type_part = ",\n                'artist_type', artists.artist_type"
1702        return f"""(SELECT JSON_GROUP_ARRAY(
1703            json_object(
1704                'item_id', artists.item_id,
1705                'name', artists.name,
1706                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1707            )) FROM artists
1708            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1709            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1710
1711    def _summary_base_columns(self) -> str:
1712        """Return the SELECT columns shared by every summary query."""
1713        # the search/sort/statistics columns are selected so ORDER BY (see sort_keys)
1714        # resolves them from the result set, like the full query's SELECT * does
1715        return f"""
1716            {self.db_table}.item_id,
1717            {self.db_table}.name,
1718            {self.db_table}.sort_name,
1719            {self.db_table}.favorite,
1720            {self.db_table}.search_name AS search_name,
1721            {self.db_table}.search_sort_name AS search_sort_name,
1722            {self.db_table}.play_count AS play_count,
1723            {self.db_table}.last_played AS last_played,
1724            {self.db_table}.timestamp_added AS timestamp_added,
1725            {self.db_table}.timestamp_modified AS timestamp_modified,
1726            json_extract({self.db_table}.metadata, '$.images') AS images,
1727            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1728
1729    async def _localized_search_fallback(
1730        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1731    ) -> list[ItemCls]:
1732        """
1733        Retry a library search using the canonical names behind a localized query.
1734
1735        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1736        canonical (English) names of matching localized items and search those, so an item is
1737        findable by the localized name the user sees. The caller's other filters (favorite,
1738        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1739        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1740        ``TranslationController.reverse_lookup_media_names``.
1741        """
1742        seen: set[Any] = set()
1743        merged: list[ItemCls] = []
1744        # iterate the canonical names in a stable order, and fetch each from the start so the
1745        # offset/limit window can be applied to the merged, de-duplicated result set
1746        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1747            for item in await self.library_items(
1748                search=name,
1749                limit=limit + offset,
1750                offset=0,
1751                _localized_fallback=False,
1752                **call_kwargs,
1753            ):
1754                if item.item_id not in seen:
1755                    seen.add(item.item_id)
1756                    merged.append(item)
1757        return merged[offset : offset + limit]
1758
1759    @abstractmethod
1760    async def _add_library_item(
1761        self,
1762        item: ItemCls,
1763        overwrite_existing: bool = False,
1764    ) -> int:
1765        """Add item to library and return the database id."""
1766
1767    @abstractmethod
1768    async def _update_library_item(
1769        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1770    ) -> None:
1771        """Update existing library record in the database."""
1772
1773    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1774        """Return the SQL WHERE clause fragment used for search filtering."""
1775        return search_name_match_clause(self.db_table, search, "search", query_params)
1776
1777    @final
1778    def _preprocess_search(self, search: str | None) -> str | None:
1779        """Normalize the search string for use in the search filter clauses."""
1780        return create_safe_string(search, True, True) if search else search
1781
1782    @final
1783    @staticmethod
1784    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1785        if genre_ids is None:
1786            return None
1787        if isinstance(genre_ids, list):
1788            normalized = [int(x) for x in genre_ids]
1789        else:
1790            normalized = [int(genre_ids)]
1791        return normalized or None
1792
1793    @final
1794    @staticmethod
1795    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1796        """Clean the query parts list by removing duplicate where statements."""
1797        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1798
1799    @final
1800    def _apply_random_subquery(  # noqa: PLR0913
1801        self,
1802        query_parts: list[str],
1803        query_params: dict[str, Any],
1804        join_parts: list[str],
1805        favorite: bool | None,
1806        search: str | None,
1807        genre_ids: list[int] | None,
1808        provider_filter: list[str] | None,
1809        played_only: bool = False,
1810        limit: int = 500,
1811        in_library_only: bool = False,
1812        reachable_via: list[str] | None = None,
1813    ) -> None:
1814        """Build a fast random subquery with all filters applied."""
1815        sub_query_parts = query_parts.copy()
1816        sub_join_parts = join_parts.copy()
1817
1818        # Apply all filters to the subquery
1819        self._apply_filters(
1820            query_parts=sub_query_parts,
1821            query_params=query_params,
1822            favorite=favorite,
1823            search=search,
1824            genre_ids=genre_ids,
1825            provider_filter=provider_filter,
1826            played_only=played_only,
1827            in_library_only=in_library_only,
1828            reachable_via=reachable_via,
1829        )
1830
1831        # Build the subquery
1832        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1833
1834        if sub_join_parts:
1835            sub_query += f" {' '.join(sub_join_parts)}"
1836
1837        if sub_query_parts:
1838            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1839
1840        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1841
1842        # The query now only consists of the random subquery, which applies all filters
1843        # within itself
1844        query_parts.clear()
1845        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1846        join_parts.clear()
1847
1848    @final
1849    def _apply_filters(
1850        self,
1851        query_parts: list[str],
1852        query_params: dict[str, Any],
1853        favorite: bool | None,
1854        search: str | None,
1855        genre_ids: list[int] | None,
1856        provider_filter: list[str] | None,
1857        played_only: bool = False,
1858        in_library_only: bool = False,
1859        reachable_via: list[str] | None = None,
1860    ) -> None:
1861        """Apply search, favorite, and provider filters."""
1862        # handle search
1863        if search:
1864            query_parts.append(self._search_filter_clause(search, query_params))
1865        # handle favorite filter
1866        if favorite is not None:
1867            query_parts.append(f"{self.db_table}.favorite = :favorite")
1868            query_params["favorite"] = favorite
1869        # handle played_only filter
1870        if played_only:
1871            query_parts.append(f"{self.db_table}.last_played > 0")
1872        # handle genre filter
1873        if genre_ids:
1874            query_params["genre_ids"] = genre_ids
1875            query_params["genre_media_type"] = self.media_type.value
1876            query_parts.append(
1877                f"EXISTS("
1878                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1879                f"WHERE gm.media_id = {self.db_table}.item_id "
1880                "AND gm.media_type = :genre_media_type "
1881                "AND gm.genre_id IN :genre_ids)"
1882            )
1883        # Apply the provider filter
1884        if provider_filter or in_library_only:
1885            query_parts.append(
1886                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1887            )
1888        # Apply the reachability filter, independent of the (in-library) provider filter above
1889        if reachable_via is not None:
1890            query_parts.append(self._reachability_filter_clause(query_params, reachable_via))
1891
1892    @final
1893    def _reachability_filter_clause(
1894        self, query_params: dict[str, Any], reachable_via: list[str]
1895    ) -> str:
1896        """
1897        Return the SQL clause that restricts items to those reachable via given providers.
1898
1899        Unlike `_provider_filter_clause`, this only checks that an available mapping to
1900        one of the given provider instances exists: it does not require that mapping to
1901        be in that provider's own library. This is used to answer "can this (already
1902        in-library) item be played through one of these providers", as opposed to
1903        "is this item favorited on one of these providers".
1904
1905        :param query_params: Query params dict; the clause's bound params are added to it.
1906        :param reachable_via: Only match items with an available mapping to one of these
1907            provider instances.
1908        """
1909        query_params["reachable_via_media_type"] = self.media_type.value
1910        query_params["reachable_via_providers"] = reachable_via
1911        return (
1912            f"EXISTS(SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} reachable_mappings "
1913            f"WHERE reachable_mappings.item_id = {self.db_table}.item_id "
1914            "AND reachable_mappings.media_type = :reachable_via_media_type "
1915            "AND reachable_mappings.available = 1 "
1916            "AND reachable_mappings.provider_instance IN :reachable_via_providers)"
1917        )
1918
1919    @final
1920    def _provider_filter_clause(
1921        self,
1922        query_params: dict[str, Any],
1923        provider_filter: list[str] | None,
1924        in_library_only: bool = False,
1925    ) -> str:
1926        """
1927        Return the SQL clause that restricts items by their provider mappings.
1928
1929        At least one of provider_filter/in_library_only must be set, otherwise the
1930        returned clause only asserts that the item has any mapping at all.
1931
1932        :param query_params: Query params dict; the clause's bound params are added to it.
1933        :param provider_filter: Only match items mapped to one of these provider instances.
1934        :param in_library_only: Only match provider mappings that are in the provider's library.
1935        """
1936        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1937        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1938        # sort index instead of materializing/sorting the whole (deduped) result set.
1939        query_params["provider_media_type"] = self.media_type.value
1940        conditions = [
1941            f"provider_mappings.item_id = {self.db_table}.item_id",
1942            "provider_mappings.media_type = :provider_media_type",
1943        ]
1944        if in_library_only:
1945            conditions.append("provider_mappings.in_library = 1")
1946        if provider_filter:
1947            provider_conditions = []
1948            for idx, prov in enumerate(provider_filter):
1949                param_name = f"provider_filter_{idx}"
1950                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1951                query_params[param_name] = prov
1952            conditions.append(f"({' OR '.join(provider_conditions)})")
1953        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1954
1955    @final
1956    def _build_final_query(
1957        self,
1958        query_parts: list[str],
1959        join_parts: list[str],
1960        order_by: str | None,
1961        summary: bool = False,
1962    ) -> tuple[str, dict[str, Any]]:
1963        """Build the final SQL query string and its (base) bound query params."""
1964        sql_query, base_query_params = self.summary_query if summary else self.base_query
1965
1966        # Add joins
1967        if join_parts:
1968            sql_query += f" {' '.join(join_parts)} "
1969
1970        # Add where clauses
1971        if query_parts:
1972            # prevent duplicate where statement
1973            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1974
1975        # Add grouping (only needed when caller-provided joins can fan out rows)
1976        # and ordering. Without a GROUP BY, SQLite can stream results directly
1977        # from the sort index instead of sorting the whole result set.
1978        if join_parts:
1979            sql_query += f" GROUP BY {self.db_table}.item_id"
1980
1981        if order_by:
1982            if sort_key := SORT_KEYS.get(order_by):
1983                sql_query += f" ORDER BY {sort_key}"
1984
1985        return sql_query, base_query_params
1986
1987    @final
1988    @staticmethod
1989    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1990        """Parse raw db Mapping into a dict."""
1991        db_row_dict = dict(db_row)
1992        db_row_dict["provider"] = "library"
1993        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1994        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1995        db_row_dict["date_added"] = datetime.fromtimestamp(
1996            db_row_dict["timestamp_added"], tz=UTC
1997        ).isoformat()
1998
1999        for key in JSON_KEYS:
2000            if key not in db_row_dict:
2001                continue
2002            if not (raw_value := db_row_dict[key]):
2003                continue
2004            db_row_dict[key] = json_loads(raw_value)
2005
2006        # parse "fully_played" as bool if present in the row
2007        if "fully_played" in db_row_dict:
2008            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
2009
2010        # copy track_album --> album
2011        if track_album := db_row_dict.get("track_album"):
2012            db_row_dict["album"] = track_album
2013            db_row_dict["disc_number"] = track_album["disc_number"]
2014            db_row_dict["track_number"] = track_album["track_number"]
2015            # always prefer album image over track image
2016            if (album_images := track_album.get("images")) and (
2017                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
2018            ):
2019                # copy album image to itemmapping single image (on the track)
2020                db_row_dict["image"] = album_thumb
2021                # also set image on the album dict for ItemMapping compatibility
2022                track_album["image"] = album_thumb
2023                if db_row_dict["metadata"].get("images"):
2024                    # merge album image with existing images
2025                    db_row_dict["metadata"]["images"] = [
2026                        album_thumb,
2027                        *db_row_dict["metadata"]["images"],
2028                    ]
2029                else:
2030                    db_row_dict["metadata"]["images"] = [album_thumb]
2031
2032        if audiobook_artists := db_row_dict.get("audiobook_artists"):
2033            _narrators = []
2034            _authors = []
2035            for artist in audiobook_artists:
2036                artist_type = artist.get("artist_type")
2037                if artist_type == "author":
2038                    _authors.append(artist)
2039                elif artist_type == "narrator":
2040                    _narrators.append(artist)
2041            if _authors:
2042                # prevent overwriting string values
2043                db_row_dict["authors"] = _authors
2044            if _narrators:
2045                # prevent overwriting string values
2046                db_row_dict["narrators"] = _narrators
2047
2048        return db_row_dict
2049
2050    @final
2051    def _ensure_provider_filter(
2052        self,
2053        provider: str | list[str] | None,
2054    ) -> list[str] | None:
2055        """Ensure the provider filter respects the current user's provider filter."""
2056        # Apply user provider filter if needed
2057        user = get_current_user()
2058        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2059        final_provider_filter: list[str] | None = None
2060        if user_provider_filter:
2061            plugin_provider_instances = {
2062                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
2063            }
2064            # User has a provider filter set
2065            if provider:
2066                # Explicit provider filter provided - validate against user's allowed providers
2067                requested_providers = [provider] if isinstance(provider, str) else provider
2068                # Only restrict access to music providers.
2069                final_provider_filter = [
2070                    p
2071                    for p in requested_providers
2072                    if p in user_provider_filter or p in plugin_provider_instances
2073                ]
2074                if not final_provider_filter:
2075                    # No overlap - user requested providers they don't have access to
2076                    raise InsufficientPermissions(
2077                        "User does not have permission to access the requested provider(s)."
2078                    )
2079            else:
2080                # No explicit filter - apply user music provider filter but keep plugin providers.
2081                final_provider_filter = list(
2082                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
2083                )
2084        elif provider is not None:
2085            # No user filter - use the provided filter as is
2086            final_provider_filter = [provider] if isinstance(provider, str) else provider
2087        return final_provider_filter
2088
2089    @final
2090    def _resolve_reachable_via(self, reachable_via: list[str] | None) -> list[str] | None:
2091        """
2092        Resolve a `reachable_via` filter against currently loaded, user-allowed providers.
2093
2094        :param reachable_via: Requested provider instance ids, or None for no filter.
2095        :return: None if no filter should be applied. Otherwise, the subset of
2096            `reachable_via` that is currently active and allowed for the current user
2097            (per `MusicController.get_active_provider_instances`). An empty list means
2098            the filter cannot match anything; callers must then return no items rather
2099            than issue a query.
2100        """
2101        if reachable_via is None:
2102            return None
2103        if not reachable_via:
2104            return []
2105        allowed_providers = set(self.mass.music.get_active_provider_instances())
2106        return [p for p in reachable_via if p in allowed_providers]
2107
2108    @final
2109    def _provider_filter_considering_reachability(
2110        self,
2111        provider: str | list[str] | None,
2112        resolved_reachable_via: list[str] | None,
2113    ) -> list[str] | None:
2114        """
2115        Resolve the `provider` filter, deferring to an active `reachable_via` filter.
2116
2117        The current user's provider access is already enforced on `resolved_reachable_via`
2118        by `_resolve_reachable_via`. So when `reachable_via` is active and no explicit
2119        `provider` filter was requested, skip `_ensure_provider_filter`'s implicit
2120        injection of the user's provider filter: that would additionally require the
2121        item's in-library mapping itself to be on one of those providers, which is
2122        stricter than (and redundant with) what `reachable_via` already checks.
2123
2124        :param provider: The explicit provider filter, as passed to `library_items`.
2125        :param resolved_reachable_via: The already-resolved `reachable_via` filter (the
2126            return value of `_resolve_reachable_via`), or None if not active.
2127        """
2128        if resolved_reachable_via is not None and provider is None:
2129            return None
2130        return self._ensure_provider_filter(provider)
2131
2132    @final
2133    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
2134        """Select the correct provider id to use for fetching the item."""
2135        if not library_item.provider_mappings:
2136            msg = (
2137                f"{self.media_type.value} {library_item.item_id} "
2138                "is no longer available on any provider"
2139            )
2140            raise MediaNotFoundError(msg)
2141        user = get_current_user()
2142        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2143        if not user_provider_filter:
2144            mapping = next(iter(library_item.provider_mappings))
2145            return (mapping.provider_instance, mapping.item_id)
2146
2147        # First prefer music provider mappings that are explicitly allowed for this user.
2148        # prefer user provider filter if available
2149        for mapping in library_item.provider_mappings:
2150            provider = self.mass.get_provider(mapping.provider_instance)
2151            if provider and provider.type == ProviderType.MUSIC:
2152                if mapping.provider_instance in user_provider_filter:
2153                    return (mapping.provider_instance, mapping.item_id)
2154
2155        # If no allowed music mapping exists, fall back to plugin mappings.
2156        for mapping in library_item.provider_mappings:
2157            provider = self.mass.get_provider(mapping.provider_instance)
2158            if provider and provider.type == ProviderType.PLUGIN:
2159                return (mapping.provider_instance, mapping.item_id)
2160
2161        # As a final fallback, preserve previous behavior.
2162        for mapping in library_item.provider_mappings:
2163            if mapping.provider_instance in user_provider_filter:
2164                return (mapping.provider_instance, mapping.item_id)
2165
2166        # fallback to first mapping
2167        mapping = next(iter(library_item.provider_mappings))
2168        return (mapping.provider_instance, mapping.item_id)
2169
2170    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
2171        """
2172        Remove images belonging to a provider from a library item's stored metadata.
2173
2174        :param db_id: The library (database) id of the item.
2175        :param provider_instance_id: The provider instance whose images should be removed.
2176        :return: True if any images were removed and the db record was updated.
2177        """
2178        # read the raw metadata straight from the db (instead of via get_library_item)
2179        # to avoid persisting any images that are only injected at read time (such as
2180        # the album thumb that gets merged into a track's images)
2181        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
2182        if not db_row or not (raw_metadata := db_row["metadata"]):
2183            return False
2184        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
2185        if not metadata.images:
2186            return False
2187        remaining = UniqueList(
2188            img for img in metadata.images if img.provider != provider_instance_id
2189        )
2190        if len(remaining) == len(metadata.images):
2191            # nothing belonged to this provider
2192            return False
2193        metadata.images = remaining or None
2194        await self.mass.music.database.update(
2195            self.db_table,
2196            {"item_id": db_id},
2197            {"metadata": serialize_to_json(metadata)},
2198        )
2199        return True
2200
2201    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
2202        """
2203        Return extra (columns, joins, params) for this media type's sync-details query.
2204
2205        Override in a subclass to select additional lightweight columns needed by the
2206        library sync change detection for this media type.
2207        """
2208        return "", "", {}
2209
2210    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
2211        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
2212        return LibraryItemSyncDetails(
2213            item_id=db_row["item_id"],
2214            favorite=bool(db_row["favorite"]),
2215            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
2216            provider_mappings=self._parse_sync_details_mappings(db_row),
2217        )
2218
2219    @final
2220    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2221        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
2222        return {
2223            ProviderMapping(
2224                item_id=raw_mapping["item_id"],
2225                provider_domain=raw_mapping["provider_domain"],
2226                provider_instance=raw_mapping["provider_instance"],
2227                available=bool(raw_mapping["available"]),
2228                in_library=parse_optional_bool(raw_mapping["in_library"]),
2229                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
2230            )
2231            for raw_mapping in json_loads(db_row["provider_mappings"])
2232        }
2233
2234    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
2235        """
2236        Parse a raw summary db row into a summary item of this controller's media type.
2237
2238        Override in a subclass to fill additional per-type fields (selected by the
2239        subclass's summary_query).
2240        """
2241        provider_mappings = self._parse_summary_provider_mappings(db_row)
2242        return self.summary_item_cls(
2243            item_id=str(db_row["item_id"]),
2244            provider="library",
2245            name=db_row["name"],
2246            sort_name=db_row["sort_name"],
2247            favorite=bool(db_row["favorite"]),
2248            provider_mappings=provider_mappings,
2249            available=self._summary_available(provider_mappings),
2250            metadata=self._parse_summary_metadata(db_row),
2251        )
2252
2253    @final
2254    @staticmethod
2255    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2256        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
2257        if not (raw_mappings := db_row["provider_mappings"]):
2258            return set()
2259        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
2260
2261    @final
2262    @staticmethod
2263    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
2264        """Compute the availability flag from a summary item's provider mappings."""
2265        # same semantics as the MediaItem.available property
2266        if not (available_providers := get_global_cache_value("available_providers")):
2267            return any(x.available for x in provider_mappings)
2268        if TYPE_CHECKING:
2269            available_providers = cast("set[str]", available_providers)
2270        return any(
2271            x.available and x.provider_instance in available_providers for x in provider_mappings
2272        )
2273
2274    @final
2275    @staticmethod
2276    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
2277        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
2278        thumb: MediaItemImage | None = None
2279        if raw_images := db_row["images"]:
2280            for image in json_loads(raw_images):
2281                if image["type"] != ImageType.THUMB.value:
2282                    continue
2283                thumb = MediaItemImage(
2284                    type=ImageType.THUMB,
2285                    path=image["path"],
2286                    provider=image["provider"],
2287                    remotely_accessible=image.get("remotely_accessible", False),
2288                )
2289                break
2290        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2291
2292    @final
2293    def _parse_summary_artist_mappings(
2294        self, db_row: Mapping[str, Any]
2295    ) -> UniqueList[ItemMappingSummary]:
2296        """Parse the aggregated slim artist mapping rows of a summary db row."""
2297        return UniqueList(
2298            ItemMappingSummary(
2299                media_type=MediaType.ARTIST,
2300                item_id=str(raw_mapping["item_id"]),
2301                provider="library",
2302                name=raw_mapping["name"],
2303                sort_name=raw_mapping["sort_name"],
2304            )
2305            for raw_mapping in json_loads(db_row["artists"])
2306        )
2307
2308    async def _adapt_query_for_collections(
2309        self,
2310        sql_query: str,
2311        query_params: dict[str, Any],
2312        summary: bool,
2313        order_by: str | None,
2314        collection_name: str | None = None,
2315        search: str | None = None,
2316    ) -> str:
2317        cache_key_json_object = f"collection_{self.api_base}"
2318        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2319        if json_object is None:
2320            # get column names of base query
2321            db_rows = await self.mass.music.database.get_rows_from_query(
2322                sql_query, query_params, limit=1, offset=0
2323            )
2324            # create a sql json_object which queries all these columns
2325            if db_rows:
2326                json_object = (
2327                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2328                )
2329                await self.mass.cache.set(
2330                    key=cache_key_json_object, category=int(summary), data=json_object
2331                )
2332            else:
2333                json_object = "json_object()"
2334
2335        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2336
2337        supported_order_keys = [
2338            "name",
2339            "name_desc",
2340            "sort_name",
2341            "sort_name_desc",
2342            "timestamp_added",
2343            "timestamp_added_desc",
2344            "timestamp_modified",
2345            "timestamp_modified_desc",
2346            "last_played",
2347            "last_played_desc",
2348            "play_count",
2349            "play_count_desc",
2350        ]
2351
2352        # additional order options subject to media type
2353        # single is targeting a single media item, collection the aggregated ones
2354        single_extra_order_keys = ""
2355        collection_extra_order_keys = ""
2356        if MediaType.AUDIOBOOK.value in self.api_base:
2357            single_extra_order_keys = "duration,"
2358            collection_extra_order_keys = "SUM(duration) as duration,"
2359            supported_order_keys += ["duration", "duration_desc"]
2360
2361        sql_query = f"""
2362        SELECT * FROM (
2363
2364            WITH
2365                joined_table as ({sql_query}),
2366                collection_extract as (
2367                    SELECT
2368                        name as media_name,
2369                        timestamp_added,
2370                        timestamp_modified,
2371                        last_played,
2372                        play_count,
2373                        {single_extra_order_keys}
2374                        json_extract(iter_coll.value, '$.title') as collection_title,
2375                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2376                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2377                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2378                        CASE
2379                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2380                            THEN 1
2381                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2382                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2383                            THEN CASE
2384                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2385                                    IN ('integer', 'real')
2386                                THEN 1
2387                                ELSE 0
2388                            END
2389                            ELSE 0
2390                        END as collection_sequence_is_numeric,
2391                        {json_object} as media_data
2392                    FROM (
2393                        SELECT * FROM joined_table
2394                    ), json_each({collections_column}) as iter_coll
2395                )
2396            SELECT
2397                'collection' as type,
2398                collection_title as name,
2399                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2400                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2401                MAX(timestamp_added) as timestamp_added,
2402                MAX(timestamp_modified) as timestamp_modified,
2403                MAX(last_played) as last_played,
2404                SUM(play_count) as play_count,
2405                {collection_extra_order_keys}
2406                json_group_array(media_data) as media_data
2407            FROM (
2408                SELECT * FROM collection_extract
2409                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2410                -- Confirmed working with sqlite 3.40.1 & 3.53
2411                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2412                ORDER BY collection_title,
2413                -- null case
2414                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2415                -- numeric before text
2416                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2417                -- order NUMERIC
2418                CASE WHEN collection_sequence_is_numeric
2419                    THEN CAST(collection_sequence AS REAL)
2420                END,
2421                -- order TEXT
2422                CASE WHEN NOT collection_sequence_is_numeric
2423                    THEN collection_sequence
2424                END COLLATE NOCASE,
2425                -- order by media name if no sequence given
2426                CASE
2427                    WHEN collection_sequence IS NULL
2428                    THEN media_name
2429                END COLLATE NOCASE
2430            )
2431            GROUP BY collection_title
2432
2433            UNION ALL
2434
2435            SELECT 'single', name, search_name, search_sort_name,
2436                timestamp_added, timestamp_modified, last_played, play_count,
2437                {single_extra_order_keys}
2438                {json_object} FROM joined_table
2439                WHERE {collections_column} IS NULL
2440                    OR {collections_column} = '[]'
2441        )
2442        """
2443
2444        if collection_name:
2445            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2446            return sql_query
2447
2448        if search:
2449            sql_query += " WHERE search_name LIKE :search"
2450
2451        if order_by:
2452            if order_by not in supported_order_keys:
2453                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2454                order_by = "name"  # fallback
2455            if sort_key := SORT_KEYS.get(order_by):
2456                sql_query += f" ORDER BY {sort_key}"
2457
2458        return sql_query
2459
2460    async def _merge_library_items(self, target_id: int, source_id: int) -> tuple[ItemCls, ItemCls]:
2461        """Merge the source library item into the target while the controller lock is held."""
2462        target_item = await self.get_library_item(target_id)
2463        source_item = await self.get_library_item(source_id)
2464        await self._validate_library_item_merge(target_item, source_item)
2465        target_row = await self.mass.music.database.get_row(self.db_table, {"item_id": target_id})
2466        source_row = await self.mass.music.database.get_row(self.db_table, {"item_id": source_id})
2467        assert target_row is not None
2468        assert source_row is not None
2469        timestamps_added = tuple(
2470            timestamp
2471            for timestamp in (
2472                int(target_row["timestamp_added"] or 0),
2473                int(source_row["timestamp_added"] or 0),
2474            )
2475            if timestamp
2476        )
2477
2478        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2479        try:
2480            source_mappings = source_item.provider_mappings
2481            source_item.provider_mappings = set()
2482            try:
2483                await self._update_library_item_for_merge(target_id, source_item)
2484            finally:
2485                source_item.provider_mappings = source_mappings
2486
2487            await self.mass.music.database.execute_write(
2488                f"""
2489                UPDATE {self.db_table}
2490                SET play_count = CASE item_id
2491                    WHEN :target_id THEN :merged_play_count
2492                    WHEN :source_id THEN 0
2493                END
2494                WHERE item_id IN (:target_id, :source_id)
2495                """,
2496                {
2497                    "target_id": target_id,
2498                    "source_id": source_id,
2499                    "merged_play_count": int(target_row["play_count"] or 0)
2500                    + int(source_row["play_count"] or 0),
2501                },
2502            )
2503            await self.mass.music.database.update(
2504                self.db_table,
2505                {"item_id": target_id},
2506                {
2507                    "favorite": bool(target_row["favorite"]) or bool(source_row["favorite"]),
2508                    "last_played": max(
2509                        int(target_row["last_played"] or 0), int(source_row["last_played"] or 0)
2510                    ),
2511                    "timestamp_added": min(timestamps_added) if timestamps_added else 0,
2512                },
2513            )
2514            await self._merge_genre_mappings(target_id, source_id)
2515            await self._merge_library_item_references(target_id, source_id)
2516            await self._merge_library_playlog(target_id, source_id)
2517            # the transfer commits in steps (see `deferred_commit`), so it is ordered to
2518            # leave the source repairable wherever it is cut short: relations are copied
2519            # rather than moved, and only dropped once the target holds them and the
2520            # provider mappings. A source that kept its relations stays a duplicate the
2521            # reconciliation pass can finish; one that lost its mappings is cleaned up.
2522            await self._copy_library_item_relations(target_id, source_id)
2523            await self.mass.music.database.execute_write(
2524                f"UPDATE {DB_TABLE_PROVIDER_MAPPINGS} SET item_id = :target_id "
2525                "WHERE media_type = :media_type AND item_id = :source_id",
2526                {
2527                    "target_id": target_id,
2528                    "source_id": source_id,
2529                    "media_type": self.media_type.value,
2530                },
2531            )
2532            await self._drop_library_item_relations(source_id)
2533            await MediaControllerBase.remove_item_from_library(self, source_id, recursive=False)
2534            merged_item = await self.get_library_item(target_id)
2535        finally:
2536            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2537
2538        return source_item, merged_item
2539
2540    async def _merge_library_items_batched(self, target_id: int, source_id: int) -> ItemCls:
2541        """Merge library items while batching the transfer's database writes."""
2542        async with self.mass.music.database.deferred_commit():
2543            source_item, merged_item = await self._merge_library_items(target_id, source_id)
2544        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
2545            self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, source_item.uri, source_item)
2546            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, merged_item.uri, merged_item)
2547        return merged_item
2548
2549    async def _validate_library_item_merge(self, target: ItemCls, source: ItemCls) -> None:
2550        """Validate that the target and source items can be merged."""
2551        if target.media_type != self.media_type or source.media_type != self.media_type:
2552            msg = "Library items must have the controller's media type"
2553            raise InvalidDataError(msg)
2554
2555    async def _update_library_item_for_merge(self, item_id: int, update: ItemCls) -> None:
2556        """Merge model state into an existing library item."""
2557        await self._update_library_item(item_id, update)
2558
2559    async def _copy_library_item_relations(self, target_id: int, source_id: int) -> None:
2560        """Copy the relations that reference the merged media item onto the target."""
2561        for table, item_column in self._library_item_relations():
2562            columns = RELATION_TABLE_COLUMNS[table]
2563            selected = ", ".join(
2564                ":target_id" if column == item_column else column for column in columns
2565            )
2566            await self.mass.music.database.execute_write(
2567                f"INSERT OR IGNORE INTO {table}({', '.join(columns)}) "
2568                f"SELECT {selected} FROM {table} WHERE {item_column} = :source_id",
2569                {"target_id": target_id, "source_id": source_id},
2570            )
2571
2572    async def _drop_library_item_relations(self, source_id: int) -> None:
2573        """Drop the relations of a merged media item once the target holds them."""
2574        for table, item_column in self._library_item_relations():
2575            await self.mass.music.database.delete(table, {item_column: source_id})
2576
2577    def _library_item_relations(self) -> tuple[tuple[str, str], ...]:
2578        """Return the (table, column) pairs holding relations to this controller's items."""
2579        if self.media_type == MediaType.ALBUM:
2580            return (
2581                (DB_TABLE_ALBUM_ARTISTS, "album_id"),
2582                (DB_TABLE_ALBUM_TRACKS, "album_id"),
2583            )
2584        if self.media_type == MediaType.ARTIST:
2585            return (
2586                (DB_TABLE_ALBUM_ARTISTS, "artist_id"),
2587                (DB_TABLE_AUDIOBOOK_ARTISTS, "artist_id"),
2588                (DB_TABLE_TRACK_ARTISTS, "artist_id"),
2589            )
2590        if self.media_type == MediaType.AUDIOBOOK:
2591            return ((DB_TABLE_AUDIOBOOK_ARTISTS, "audiobook_id"),)
2592        if self.media_type == MediaType.TRACK:
2593            return (
2594                (DB_TABLE_ALBUM_TRACKS, "track_id"),
2595                (DB_TABLE_TRACK_ARTISTS, "track_id"),
2596            )
2597        return ()
2598
2599    async def _merge_library_item_references(self, target_id: int, source_id: int) -> None:
2600        """Transfer references to the source item owned by specialized controllers."""
2601        return
2602
2603    async def _merge_genre_mappings(self, target_id: int, source_id: int) -> None:
2604        """Transfer genre mappings and exclusions to the target item."""
2605        values = {
2606            "target_id": target_id,
2607            "source_id": source_id,
2608            "media_type": self.media_type.value,
2609        }
2610        await self.mass.music.database.execute_write(
2611            f"""
2612            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2613                genre_id, media_id, media_type, alias, is_derived, is_manual
2614            )
2615            SELECT genre_id, :target_id, media_type, alias, is_derived, is_manual
2616            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2617            WHERE media_id = :source_id AND media_type = :media_type
2618            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2619                alias = CASE
2620                    WHEN excluded.is_manual AND NOT is_manual
2621                    THEN COALESCE(excluded.alias, alias)
2622                    ELSE COALESCE(alias, excluded.alias)
2623                END,
2624                is_derived = is_derived OR excluded.is_derived,
2625                is_manual = is_manual OR excluded.is_manual
2626            """,
2627            values,
2628        )
2629        await self.mass.music.database.execute_write(
2630            f"""
2631            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2632                genre_id, media_id, media_type
2633            )
2634            SELECT genre_id, :target_id, media_type
2635            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2636            WHERE media_id = :source_id AND media_type = :media_type
2637            """,
2638            values,
2639        )
2640        await self.mass.music.database.delete(
2641            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
2642            {"media_id": source_id, "media_type": self.media_type.value},
2643        )
2644        await self.mass.music.database.delete(
2645            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
2646            {"media_id": source_id, "media_type": self.media_type.value},
2647        )
2648
2649    async def _merge_genre_references(self, target_id: int, source_id: int) -> None:
2650        """Transfer media mappings and exclusions that point to the source genre."""
2651        values = {"target_id": target_id, "source_id": source_id}
2652        await self.mass.music.database.execute_write(
2653            f"""
2654            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2655                genre_id, media_id, media_type, alias, is_derived, is_manual
2656            )
2657            SELECT :target_id, media_id, media_type, alias, is_derived, is_manual
2658            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2659            WHERE genre_id = :source_id
2660            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2661                alias = CASE
2662                    WHEN excluded.is_manual AND NOT is_manual
2663                    THEN COALESCE(excluded.alias, alias)
2664                    ELSE COALESCE(alias, excluded.alias)
2665                END,
2666                is_derived = is_derived OR excluded.is_derived,
2667                is_manual = is_manual OR excluded.is_manual
2668            """,
2669            values,
2670        )
2671        await self.mass.music.database.execute_write(
2672            f"""
2673            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2674                genre_id, media_id, media_type
2675            )
2676            SELECT :target_id, media_id, media_type
2677            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2678            WHERE genre_id = :source_id
2679            """,
2680            values,
2681        )
2682        await self.mass.music.database.delete(
2683            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING, {"genre_id": source_id}
2684        )
2685        await self.mass.music.database.delete(
2686            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION, {"genre_id": source_id}
2687        )
2688
2689    async def _merge_library_playlog(self, target_id: int, source_id: int) -> None:
2690        """Transfer library-keyed playlog rows using the normal latest-entry semantics."""
2691        values = {
2692            "target_id": target_id,
2693            "source_id": source_id,
2694            "media_type": self.media_type.value,
2695        }
2696        await self.mass.music.database.execute_write(
2697            f"""
2698            INSERT INTO {DB_TABLE_PLAYLOG}(
2699                item_id, provider, media_type, name, image, artists, timestamp,
2700                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2701            )
2702            SELECT
2703                :target_id, provider, media_type, name, image, artists, timestamp,
2704                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2705            FROM {DB_TABLE_PLAYLOG}
2706            WHERE item_id = :source_id AND provider = 'library' AND media_type = :media_type
2707            ON CONFLICT(item_id, provider, media_type, userid) DO UPDATE SET
2708                name = CASE WHEN excluded.timestamp > timestamp THEN excluded.name ELSE name END,
2709                image = CASE WHEN excluded.timestamp > timestamp THEN excluded.image ELSE image END,
2710                artists = CASE WHEN excluded.timestamp > timestamp THEN excluded.artists ELSE artists END,
2711                timestamp = MAX(timestamp, excluded.timestamp),
2712                fully_played = CASE
2713                    WHEN excluded.timestamp > timestamp THEN excluded.fully_played ELSE fully_played
2714                END,
2715                seconds_played = CASE
2716                    WHEN excluded.timestamp > timestamp THEN excluded.seconds_played
2717                    ELSE seconds_played
2718                END,
2719                queue_id = CASE
2720                    WHEN excluded.timestamp > timestamp THEN excluded.queue_id ELSE queue_id
2721                END,
2722                user_initiated = user_initiated OR excluded.user_initiated,
2723                playback_speed = CASE
2724                    WHEN excluded.timestamp > timestamp THEN excluded.playback_speed
2725                    ELSE playback_speed
2726                END
2727            """,
2728            values,
2729        )
2730        await self.mass.music.database.delete(
2731            DB_TABLE_PLAYLOG,
2732            {
2733                "item_id": source_id,
2734                "provider": "library",
2735                "media_type": self.media_type.value,
2736            },
2737        )
2738