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