/
/
/
1"""Cached-payload mixin for providers whose recommendations come from one bulk payload."""
2
3from __future__ import annotations
4
5import asyncio
6import dataclasses
7import logging
8import time
9from typing import TYPE_CHECKING, TypeVar, cast
10
11from music_assistant_models.media_items import RecommendationFolder
12from music_assistant_models.unique_list import UniqueList
13
14from music_assistant.helpers.util import join_task
15
16if TYPE_CHECKING:
17 from collections.abc import Coroutine
18 from typing import Any
19
20 from music_assistant_models.media_items import BrowseFolder, ItemMapping, MediaItemType
21
22 from music_assistant.helpers.json import SerializableType
23
24 # type the mixin against the Provider base it composes with, so self.mass,
25 # self.logger, instance_id and the unload() chain resolve without redeclaring
26 # any (final) Provider members; at runtime the mixin stays a plain object base
27 from music_assistant.models.provider import Provider as _MixinBase
28else:
29 _MixinBase = object
30
31_T = TypeVar("_T")
32
33# Key of the persistent payload cache entry (stored with provider=instance_id).
34# Kept identical to the key the previous @use_cache-based implementation produced,
35# so payloads persisted before the redesign still warm cold instances.
36_PAYLOAD_CACHE_KEY = "_cached_recommendation_payload"
37
38
39class RecommendationPayloadMixin(_MixinBase):
40 """
41 Mixin serving recommendation rows and items from a single (cached) bulk payload.
42
43 The provider implements _fetch_recommendation_payload() with its existing bulk
44 backend fetch+parse; the mixin derives both the fast rows call and the per-row
45 items call from that payload.
46
47 Caching works in two layers:
48 - In memory: the last fetched payload is kept on the instance and served directly
49 while it is younger than recommendation_payload_ttl (no cache-db round trip).
50 A stale in-memory payload is served immediately while a single background
51 refresh replaces it (stale-while-revalidate).
52 - Persistent: every successful fetch is stored in mass.cache (persistent=True), so
53 a cold instance (e.g. after a restart) warms from the cache db with at most one
54 read; a stale persisted payload is likewise served while one refresh runs.
55
56 Concurrent callers share one in-flight fetch (single-flight); the shared fetch is
57 isolated from caller cancellation, so a timed-out caller cannot cancel it for the
58 other waiters and its result still lands in memory and cache.
59
60 All background work runs in tasks created via mass.create_task, so it is cancelled
61 on server stop; the mixin's unload() additionally cancels any in-flight fetch or
62 refresh when the provider itself is unloaded. For that override to be reachable,
63 list the mixin before the provider base class, e.g.
64 ``class MyProvider(RecommendationPayloadMixin, MusicProvider)``. A cancelled fetch
65 does not poison later calls: the next call simply starts a fresh one.
66 """
67
68 recommendation_payload_ttl: int = 3600
69 """Seconds a fetched payload is served as fresh. Subclasses may override."""
70
71 _recommendation_payload_task: asyncio.Task[list[RecommendationFolder]] | None = None
72 _recommendation_refresh_task: asyncio.Task[list[RecommendationFolder]] | None = None
73 _recommendation_payload_memory: list[RecommendationFolder] | None = None
74 _recommendation_payload_timestamp: float = 0.0
75 _recommendation_payload_cache_checked: bool = False
76
77 async def unload(self, is_removed: bool = False) -> None:
78 """
79 Handle unload/close of the provider.
80
81 Cancels any in-flight payload fetch/refresh task and awaits its completion,
82 so no payload work keeps running while the provider tears down, before
83 continuing the regular provider unload chain.
84
85 :param is_removed: True when the provider is removed from the configuration.
86 """
87 tasks = [
88 task
89 for task in (self._recommendation_payload_task, self._recommendation_refresh_task)
90 if task is not None and not task.done()
91 ]
92 for task in tasks:
93 task.cancel()
94 if tasks:
95 await asyncio.gather(*tasks, return_exceptions=True)
96 self._recommendation_payload_task = None
97 self._recommendation_refresh_task = None
98 await super().unload(is_removed)
99
100 async def _recommendation_rows_from_payload(self) -> list[RecommendationFolder]:
101 """Return all payload folders as rows, without items."""
102 # replace() keeps every payload field (image, is_playable, media_type, type, ...)
103 # instead of enumerating them, so new folder fields survive the copy automatically
104 return [
105 dataclasses.replace(folder, items=UniqueList())
106 for folder in await self._recommendation_payload()
107 ]
108
109 async def _recommendation_items_from_payload(
110 self, item_id: str
111 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
112 """
113 Return the items of the payload folder matching the given item_id.
114
115 :param item_id: The item_id of the recommendation folder (empty result if unknown).
116 """
117 for folder in await self._recommendation_payload():
118 if folder.item_id == item_id:
119 return folder.items
120 return UniqueList()
121
122 async def _recommendation_payload(self) -> list[RecommendationFolder]:
123 """Return the full recommendations payload, cached and deduplicated."""
124 if (payload := self._recommendation_payload_memory) is not None:
125 if (
126 time.monotonic() - self._recommendation_payload_timestamp
127 < self.recommendation_payload_ttl
128 ):
129 return payload
130 # stale: serve immediately while a (deduplicated) background refresh runs
131 self._schedule_recommendation_refresh()
132 return payload
133 # cold instance: single-flight load from the persistent cache or the backend
134 task = self._recommendation_payload_task
135 if task is None or task.done():
136 task = self._start_recommendation_task(
137 self._cold_load_recommendation_payload(),
138 task_id=f"recommendation_payload_fetch.{self.instance_id}",
139 )
140 self._recommendation_payload_task = task
141 # a timed-out caller must not cancel the load: it still has to complete to warm
142 # memory + cache for the other waiters
143 return await join_task(task)
144
145 async def _refresh_recommendation_payload(self) -> list[RecommendationFolder]:
146 """
147 Force-fetch a fresh payload and store it in memory and the payload cache.
148
149 Unlike _recommendation_payload, this never serves cached data: use it when the
150 cached payload is known to be outdated (e.g. after detecting rotated backend ids).
151 Subsequent _recommendation_payload calls serve the refreshed payload.
152 """
153 task = self._schedule_recommendation_refresh()
154 return await join_task(task)
155
156 def _schedule_recommendation_refresh(self) -> asyncio.Task[list[RecommendationFolder]]:
157 """Return the in-flight refresh task, starting one if none is running."""
158 task = self._recommendation_refresh_task
159 if task is None or task.done():
160 task = self._start_recommendation_task(
161 self._fetch_and_store_recommendation_payload(),
162 task_id=f"recommendation_payload_refresh.{self.instance_id}",
163 )
164 self._recommendation_refresh_task = task
165 return task
166
167 async def _cold_load_recommendation_payload(self) -> list[RecommendationFolder]:
168 """Load the payload on a cold instance: persisted cache entry or backend fetch."""
169 if not self._recommendation_payload_cache_checked:
170 data, is_fresh, found = await self.mass.cache.get_with_freshness(
171 _PAYLOAD_CACHE_KEY,
172 provider=self.instance_id,
173 base_class=RecommendationFolder,
174 include_expired=True,
175 )
176 # set after the await: a cancelled read is retried on the next cold call
177 self._recommendation_payload_cache_checked = True
178 if found and data is not None:
179 payload = cast("list[RecommendationFolder]", data)
180 self._recommendation_payload_memory = payload
181 if is_fresh:
182 self._recommendation_payload_timestamp = time.monotonic()
183 else:
184 # expired entry: serve it now, refresh it in the background
185 self._recommendation_payload_timestamp = (
186 time.monotonic() - self.recommendation_payload_ttl
187 )
188 self._schedule_recommendation_refresh()
189 return payload
190 return await self._fetch_and_store_recommendation_payload()
191
192 async def _fetch_and_store_recommendation_payload(self) -> list[RecommendationFolder]:
193 """Fetch the payload from the backend and store it in memory + persistent cache."""
194 payload = await self._fetch_recommendation_payload()
195 # memory first: warm calls are served from here without touching the cache db,
196 # and a caller landing before the background store completes still gets a hit
197 self._recommendation_payload_memory = payload
198 self._recommendation_payload_timestamp = time.monotonic()
199 self._start_recommendation_task(
200 self.mass.cache.set(
201 key=_PAYLOAD_CACHE_KEY,
202 data=cast("SerializableType", payload),
203 expiration=self.recommendation_payload_ttl,
204 provider=self.instance_id,
205 persistent=True,
206 allow_expired_cache=True,
207 )
208 )
209 return payload
210
211 def _start_recommendation_task(
212 self, coro: Coroutine[Any, Any, _T], task_id: str | None = None
213 ) -> asyncio.Task[_T]:
214 """Create a tracked background task that logs failures through the provider logger."""
215 task = self.mass.create_task(coro, task_id=task_id)
216 # always retrieve the task's exception: without a waiter (background refresh,
217 # cancelled sole caller) it would otherwise surface as a raw loop-handler error
218 task.add_done_callback(self._log_recommendation_task_failure)
219 return task
220
221 def _log_recommendation_task_failure(self, task: asyncio.Task[Any]) -> None:
222 """Log (and thereby retrieve) the exception of a finished payload task, if any."""
223 if task.cancelled():
224 return
225 if (err := task.exception()) is not None:
226 self.logger.warning(
227 "Recommendation payload task failed: %s",
228 str(err),
229 exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
230 )
231
232 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
233 """Fetch and parse the full recommendations payload (folders WITH items)."""
234 # the provider using this mixin supplies its bulk backend fetch here
235 raise NotImplementedError
236