music-assistant-server

1"""Setup flow engine: registry and API commands for interactive provider/player setup."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import time
8from dataclasses import dataclass
9from functools import partial
10from typing import TYPE_CHECKING, Any
11from uuid import uuid4
12
13from music_assistant_models.auth import Scope
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import ConfigEntryType, EventType, FlowStepType
16from music_assistant_models.errors import (
17    ActionUnavailable,
18    InsufficientPermissions,
19    SetupFailedError,
20)
21from music_assistant_models.setup_flow import SetupFlowStep
22
23from music_assistant.constants import CONF_PLAYERS, CONF_PROVIDERS
24from music_assistant.controllers.config.helpers import _AUTH_ERROR_CODES
25from music_assistant.helpers.api import api_command
26from music_assistant.helpers.util import import_module_in_thread, load_provider_module
27from music_assistant.models.player import Player
28from music_assistant.models.setup_flow import (
29    AbortFlow,
30    FlowReason,
31    SetupFlowContext,
32    SetupFlowError,
33    SetupSession,
34    StepExpiredError,
35)
36
37if TYPE_CHECKING:
38    from collections.abc import Awaitable, Callable
39
40    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
41    from music_assistant_models.provider import ProviderManifest
42
43    from music_assistant import MusicAssistant
44
45LOGGER = logging.getLogger(__name__)
46
47# a flow with no client interaction or step changes for this long is garbage collected
48IDLE_FLOW_TTL = 15 * 60
49FLOW_SWEEP_INTERVAL = 60
50# how long start/submit wait for the flow coroutine to produce the (next) step;
51# generous because finish() may install requirements and load the provider
52NEXT_STEP_TIMEOUT = 120
53# bound on waiting for a cancelled flow's cleanup (author finally-blocks)
54FLOW_ABORT_CLEANUP_TIMEOUT = 10
55
56
57@dataclass
58class ActiveSetupFlow:
59    """Registry record for a running setup flow."""
60
61    session: SetupSession
62    # target_key: identifies what the flow is (re)configuring; one flow per target
63    target_key: str
64    # required_scope: the scope the starting command required; re-checked on
65    # every flows/* continuation command
66    required_scope: Scope
67    task: asyncio.Task[None] | None = None
68
69
70class SetupFlowMixin:
71    """Mixin providing the setup flow engine for the ConfigController."""
72
73    # registry of running flows, keyed by flow_id (lazily created per instance)
74    _flows: dict[str, ActiveSetupFlow] | None = None
75    _flow_sweep_handle: asyncio.TimerHandle | None = None
76    # required scopes of recently finished flows: terminal steps can publish after
77    # the registry pop (cancel-driven aborts), and the event-scope filter still
78    # needs to resolve them; bounded FIFO
79    _finished_flow_scopes: dict[str, Scope] | None = None
80
81    # Type hints for attributes/methods provided by the class this mixin is used with
82    if TYPE_CHECKING:
83        mass: MusicAssistant
84
85        @property
86        def onboard_done(self) -> bool: ...  # noqa: D102
87
88        def get(self, key: str, default: Any = None) -> Any: ...  # noqa: D102
89
90        def set(self, key: str, value: Any, immediate: bool = False) -> None: ...  # noqa: D102
91
92        def encrypt_string(self, str_value: str) -> str: ...  # noqa: D102
93
94        def decrypt_string(self, encrypted_str: str) -> str: ...  # noqa: D102
95
96        async def get_provider_configs(  # noqa: D102
97            self, provider_type: Any = None, provider_domain: str | None = None
98        ) -> list[ProviderConfig]: ...
99
100        async def get_provider_config(self, instance_id: str) -> ProviderConfig: ...  # noqa: D102
101
102        def update_provider_last_error(self, instance_id: str, error: Any) -> None: ...  # noqa: D102
103
104        async def get_player_config(self, player_id: str) -> Any: ...  # noqa: D102
105
106        async def _create_provider_instance(
107            self,
108            provider_domain: str,
109            values: dict[str, ConfigValueType],
110            setup_data: dict[str, Any] | None = None,
111        ) -> ProviderConfig: ...
112
113    @api_command("config/providers/setup", required_scope=Scope.CONFIG_PROVIDERS_WRITE)
114    async def setup_provider(self, provider_domain: str) -> SetupFlowStep:
115        """
116        Start the setup flow to add a new instance of the given provider.
117
118        For a provider without a setup flow (no user input needed) the instance is
119        created right away and a FINISH step is returned, so the add-provider path
120        is uniform for all providers.
121
122        :param provider_domain: Domain of the provider to add an instance of.
123        """
124        for manifest in self.mass.get_provider_manifests():
125            if manifest.domain == provider_domain:
126                break
127        else:
128            msg = f"Unknown provider domain: {provider_domain}"
129            raise KeyError(msg)
130        owner = f"provider.{provider_domain}"
131        # fail fast on conditions that would otherwise only surface at save
132        existing = await self.get_provider_configs(provider_domain=provider_domain)
133        if existing and not manifest.multi_instance:
134            return self._synthesized_step(FlowStepType.ABORT, owner, reason="already_configured")
135        if manifest.depends_on:
136            dep_configs = await self.get_provider_configs(provider_domain=manifest.depends_on)
137            if not any(dep_conf.enabled for dep_conf in dep_configs):
138                return self._synthesized_step(
139                    FlowStepType.ABORT, owner, reason="missing_dependency"
140                )
141        flow_module = await self._get_setup_flow_module(manifest)
142        if flow_module is None:
143            # zero-input provider: create the instance immediately and report it
144            # as an (already) finished flow
145            config = await self._create_provider_instance(provider_domain, {})
146            return self._synthesized_step(
147                FlowStepType.FINISH, owner, result={"instance_id": config.instance_id}
148            )
149        context = SetupFlowContext(kind="setup", reason="user", domain=provider_domain)
150        return await self._start_flow(
151            flow_coro=flow_module.run_setup,
152            context=context,
153            target_key=f"provider_setup:{provider_domain}",
154            required_scope=Scope.CONFIG_PROVIDERS_WRITE,
155            finish_handler=self._finish_provider_setup,
156        )
157
158    @api_command("config/providers/reconfigure", required_scope=Scope.CONFIG_PROVIDERS_WRITE)
159    async def reconfigure_provider(self, instance_id: str) -> SetupFlowStep:
160        """
161        Start the reconfigure flow on an existing provider instance (covers reauth).
162
163        :param instance_id: The provider instance to reconfigure.
164        """
165        raw_conf = self.get(f"{CONF_PROVIDERS}/{instance_id}")
166        if not raw_conf:
167            msg = f"No config found for provider id {instance_id}"
168            raise KeyError(msg)
169        domain: str = raw_conf["domain"]
170        manifest = self.mass.get_provider_manifest(domain)
171        owner = f"provider.{domain}"
172        flow_module = await self._get_setup_flow_module(manifest)
173        if flow_module is None:
174            # flow-less providers have nothing to reconfigure;
175            # their failures are environmental (reload/retry covers them)
176            return self._synthesized_step(FlowStepType.ABORT, owner, reason="nothing_to_configure")
177        context = SetupFlowContext(
178            kind="reconfigure",
179            reason=self._reconfigure_reason(raw_conf.get("last_error")),
180            domain=domain,
181            instance_id=instance_id,
182            setup_data=self._decrypt_values(raw_conf.get("setup_data") or {}),
183            values=self._decrypt_values(raw_conf.get("values") or {}),
184        )
185        return await self._start_flow(
186            flow_coro=flow_module.run_setup,
187            context=context,
188            target_key=f"provider_reconfigure:{instance_id}",
189            required_scope=Scope.CONFIG_PROVIDERS_WRITE,
190            finish_handler=self._finish_provider_reconfigure,
191        )
192
193    @api_command("config/players/setup", required_scope=Scope.CONFIG_PLAYERS_WRITE)
194    async def setup_player(self, player_id: str) -> SetupFlowStep:
195        """
196        Start the setup flow for a player (e.g. pairing).
197
198        A player that itself needs no setup but wraps protocol child player(s) that do
199        (universal players, or native players wrapping protocol children) delegates to
200        the child's setup flow: to the single child that needs setup directly, or - when
201        more than one does - via a form that lets the user pick which child to set up.
202        The child's flow persists to the child's own config.
203
204        Also serves on-demand re-runs: when nothing needs setup (anymore), delegation
205        falls back to any child that merely has a flow, so a step the user skipped
206        earlier - an optional pairing, say - remains reachable.
207
208        :param player_id: The player to set up.
209        """
210        # deliberately no raise_unavailable: a player that needs setup is serialized
211        # as unavailable, and that is exactly the player this command targets
212        player = self.mass.players.get_player(player_id)
213        if player is None:
214            msg = f"Player {player_id} not found"
215            raise KeyError(msg)
216        owner = f"provider.{player.provider.domain}"
217        target_key = f"player_setup:{player_id}"
218        if player.implements_setup_flow:
219            # the player implements its own setup flow: run it directly
220            return await self._start_flow(
221                flow_coro=player.run_setup_flow,
222                context=self._player_flow_context(player),
223                target_key=target_key,
224                required_scope=Scope.CONFIG_PLAYERS_WRITE,
225                finish_handler=self._finish_player_setup,
226            )
227        # no direct setup: delegate to protocol child player(s), preferring the ones
228        # that actually need setup and falling back to any that can re-run their flow
229        children = self._protocol_children_with_setup_flow(player, needing_only=True)
230        if not children:
231            children = self._protocol_children_with_setup_flow(player, needing_only=False)
232        if len(children) == 1:
233            child = children[0]
234            return await self._start_flow(
235                flow_coro=child.run_setup_flow,
236                context=self._player_flow_context(child),
237                # key on the child: a direct setup of the child must replace this flow
238                target_key=f"player_setup:{child.player_id}",
239                required_scope=Scope.CONFIG_PLAYERS_WRITE,
240                finish_handler=self._finish_player_setup,
241            )
242        if children:
243            return await self._start_flow(
244                flow_coro=partial(self._run_child_selection_flow, children),
245                context=self._player_flow_context(player),
246                target_key=target_key,
247                required_scope=Scope.CONFIG_PLAYERS_WRITE,
248                finish_handler=self._finish_player_setup,
249            )
250        # nothing on this player (or its children) to configure
251        return self._synthesized_step(FlowStepType.ABORT, owner, reason="nothing_to_configure")
252
253    @api_command("config/flows/submit")
254    async def submit_setup_flow(
255        self, flow_id: str, values: dict[str, ConfigValueType]
256    ) -> SetupFlowStep:
257        """
258        Submit the user's values for the flow's pending FORM step.
259
260        Returns the flow's next step, or the same FORM step (with per-field errors
261        set) when validation failed.
262
263        :param flow_id: The id of the running flow.
264        :param values: The raw values for the form's config entries.
265        """
266        flow = self._get_flow(flow_id)
267        self._check_flow_permission(flow)
268        if (error_step := flow.session.handle_submit(values)) is not None:
269            return error_step
270        # wait (bounded) for the coroutine to produce the next step
271        submitted_step = flow.session.current_step
272        await flow.session.wait_for_step_change(NEXT_STEP_TIMEOUT)
273        step = flow.session.current_step
274        assert step is not None  # an accepted submit implies a published FORM step
275        if step is submitted_step:
276            # rare: the coroutine is still working on the next step. The submitted
277            # form's input future is already consumed, so re-serving the form would
278            # invite a doomed resubmit - publish a progress step (so flows/get agrees)
279            # and let the coroutine's next publish deliver the real step
280            flow.session.progress("working")
281            step = flow.session.current_step
282            assert step is not None
283        return step
284
285    @api_command("config/flows/get")
286    async def get_setup_flow(self, flow_id: str) -> SetupFlowStep:
287        """
288        Return the current step of a running flow (idempotent re-render, never advances).
289
290        :param flow_id: The id of the running flow.
291        """
292        flow = self._get_flow(flow_id)
293        self._check_flow_permission(flow)
294        flow.session.last_activity = time.monotonic()
295        if (step := flow.session.current_step) is None:
296            raise ActionUnavailable("The setup flow has not produced a step yet")
297        return step
298
299    @api_command("config/flows/abort")
300    async def abort_setup_flow(self, flow_id: str) -> None:
301        """
302        Abort a running flow (user cancelled).
303
304        :param flow_id: The id of the running flow.
305        """
306        flow = self._get_flow(flow_id)
307        self._check_flow_permission(flow)
308        await self._abort_flow(flow, reason="aborted")
309
310    def get_setup_flow_required_scope(self, flow_id: str) -> Scope | None:
311        """
312        Return the scope required to receive/interact with the given setup flow.
313
314        Also resolves recently finished flows (their terminal step can publish
315        just after the registry pop). Returns None when the flow is unknown.
316
317        :param flow_id: The id of the flow.
318        """
319        if flow := self._setup_flows.get(flow_id):
320            return flow.required_scope
321        if self._finished_flow_scopes:
322            return self._finished_flow_scopes.get(flow_id)
323        return None
324
325    def _pop_flow(self, flow: ActiveSetupFlow) -> None:
326        """Remove a flow from the registry, retaining its scope for late events."""
327        self._setup_flows.pop(flow.session.flow_id, None)
328        if self._finished_flow_scopes is None:
329            self._finished_flow_scopes = {}
330        finished = self._finished_flow_scopes
331        finished[flow.session.flow_id] = flow.required_scope
332        while len(finished) > 64:
333            finished.pop(next(iter(finished)))
334
335    async def _start_flow(
336        self,
337        *,
338        flow_coro: Callable[[SetupSession], Awaitable[Any]],
339        context: SetupFlowContext,
340        target_key: str,
341        required_scope: Scope,
342        finish_handler: Callable[
343            [SetupSession, dict[str, ConfigValueType]], Awaitable[dict[str, str]]
344        ],
345    ) -> SetupFlowStep:
346        """Register and start a new flow, returning its first published step."""
347        # one flow per target: starting anew replaces (aborts) a lingering previous flow.
348        # re-scan after every await: the abort yields, so a concurrent start for the same
349        # target may have registered a new flow in the meantime
350        while existing_flow := next(
351            (f for f in self._setup_flows.values() if f.target_key == target_key), None
352        ):
353            await self._abort_flow(existing_flow, reason="replaced")
354        flow_id = uuid4().hex
355        session = SetupSession(self.mass, flow_id, context, finish_handler)
356        flow = ActiveSetupFlow(
357            session=session, target_key=target_key, required_scope=required_scope
358        )
359        self._setup_flows[flow_id] = flow
360        LOGGER.debug("Starting setup flow %s for %s", flow_id, target_key)
361        flow.task = self.mass.create_task(self._run_flow(flow, flow_coro))
362        self._schedule_flow_sweep()
363        if session.current_step is None:
364            await session.wait_for_step_change(NEXT_STEP_TIMEOUT)
365        if (step := session.current_step) is None:
366            await self._abort_flow(flow, reason="internal_error")
367            raise SetupFailedError("Setup flow did not produce a first step")
368        return step
369
370    async def _run_flow(
371        self, flow: ActiveSetupFlow, flow_coro: Callable[[SetupSession], Awaitable[Any]]
372    ) -> None:
373        """Drive the flow coroutine and convert its outcome into a terminal step."""
374        session = flow.session
375        try:
376            await flow_coro(session)
377        except AbortFlow as err:
378            session.publish_abort(err.reason)
379        except StepExpiredError:
380            session.publish_abort("timed_out")
381        except SetupFlowError as err:
382            # the author did not catch a finish failure: end with the failure message
383            session.publish_abort(str(err) or "internal_error")
384        except asyncio.CancelledError:
385            # abort/replace/shutdown: the author's cleanup (finally blocks) has run;
386            # the canceller publishes the ABORT step. Never swallow the cancellation.
387            raise
388        except Exception:
389            LOGGER.exception("Unhandled error in setup flow for %s", session.context.domain)
390            session.publish_abort("internal_error")
391        else:
392            if not session.finished:
393                LOGGER.error(
394                    "Setup flow for %s returned without calling finish()", session.context.domain
395                )
396                session.publish_abort("internal_error")
397        finally:
398            LOGGER.debug("Setup flow %s ended", session.flow_id)
399            session.close()
400            self._pop_flow(flow)
401
402    async def _abort_flow(self, flow: ActiveSetupFlow, reason: str) -> None:
403        """
404        Abort the given flow with the given reason and clean it up.
405
406        Cancelling raises CancelledError inside the flow coroutine so the author's
407        cleanup (try/finally around pairing sessions etc.) runs before the terminal
408        ABORT step goes out.
409        """
410        if flow.task is not None and not flow.task.done():
411            flow.task.cancel()
412            # wait() shields us from the task's CancelledError without
413            # masking a cancellation of the caller itself; the timeout keeps a
414            # wedged author cleanup (e.g. a hanging pairing teardown) from
415            # stalling the abort and any replacement flow indefinitely
416            _, pending = await asyncio.wait([flow.task], timeout=FLOW_ABORT_CLEANUP_TIMEOUT)
417            if pending:
418                LOGGER.warning(
419                    "Setup flow for %s did not clean up within %ss after cancellation",
420                    flow.session.context.domain,
421                    FLOW_ABORT_CLEANUP_TIMEOUT,
422                )
423                # the wedged task never reaches _run_flow's finally: close the
424                # session here so the unauthenticated callback route is dropped
425                flow.session.close()
426        self._pop_flow(flow)
427        current_step = flow.session.current_step
428        if current_step is None or current_step.type not in (
429            FlowStepType.FINISH,
430            FlowStepType.ABORT,
431        ):
432            flow.session.publish_abort(reason)
433
434    async def _finish_provider_setup(
435        self, session: SetupSession, values: dict[str, ConfigValueType]
436    ) -> dict[str, str]:
437        """Finish handler for provider setup flows: create, persist and load the instance."""
438        try:
439            config = await self._create_provider_instance(
440                session.context.domain, {}, setup_data=self._encrypt_values(values)
441            )
442        except Exception as err:
443            raise SetupFlowError(
444                str(err) or err.__class__.__name__,
445                translation_key=getattr(err, "translation_key", None),
446            ) from err
447        return {"instance_id": config.instance_id}
448
449    async def _finish_provider_reconfigure(
450        self, session: SetupSession, values: dict[str, ConfigValueType]
451    ) -> dict[str, str]:
452        """Finish handler for provider reconfigure flows: merge setup_data and reload."""
453        instance_id = session.context.instance_id
454        assert instance_id is not None  # always set for reconfigure flows
455        conf_key = f"{CONF_PROVIDERS}/{instance_id}"
456        raw_conf = self.get(conf_key)
457        if not raw_conf:
458            raise SetupFlowError(f"Provider {instance_id} no longer exists")
459        snapshot = dict(raw_conf.get("setup_data") or {})
460        self.set(f"{conf_key}/setup_data", {**snapshot, **self._encrypt_values(values)})
461        try:
462            config = await self.get_provider_config(instance_id)
463            await self.mass.load_provider_config(config)
464        except asyncio.CancelledError:
465            self.set(f"{conf_key}/setup_data", snapshot)
466            raise
467        except Exception as err:
468            # reloading with the new values failed: restore the previous setup_data
469            self.set(f"{conf_key}/setup_data", snapshot)
470            raise SetupFlowError(
471                str(err) or err.__class__.__name__,
472                translation_key=getattr(err, "translation_key", None),
473            ) from err
474        self.update_provider_last_error(instance_id, None)
475        return {"instance_id": instance_id}
476
477    async def _finish_player_setup(
478        self, session: SetupSession, values: dict[str, ConfigValueType]
479    ) -> dict[str, str]:
480        """Finish handler for player setup flows: persist and apply the collected setup data."""
481        player_id = session.context.player_id
482        assert player_id is not None  # always set for player flows
483        conf_key = f"{CONF_PLAYERS}/{player_id}"
484        raw_conf = self.get(conf_key)
485        if not raw_conf:
486            raise SetupFlowError(f"No config found for player {player_id}")
487        snapshot = dict(raw_conf.get("setup_data") or {})
488        self.set(f"{conf_key}/setup_data", {**snapshot, **self._encrypt_values(values)})
489        try:
490            config = await self.get_player_config(player_id)
491            changed_keys = {f"setup_data/{key}" for key in values}
492            await self.mass.players.on_player_config_change(config, changed_keys)
493        except asyncio.CancelledError:
494            self.set(f"{conf_key}/setup_data", snapshot)
495            raise
496        except Exception as err:
497            # reading back or applying the updated config failed: restore the previous setup_data
498            self.set(f"{conf_key}/setup_data", snapshot)
499            raise SetupFlowError(
500                str(err) or err.__class__.__name__,
501                translation_key=getattr(err, "translation_key", None),
502            ) from err
503        self.mass.signal_event(EventType.PLAYER_CONFIG_UPDATED, object_id=player_id, data=config)
504        return {"player_id": player_id}
505
506    def _player_flow_context(self, player: Player) -> SetupFlowContext:
507        """Build the setup flow context (with decrypted prefill) for the given player."""
508        raw_conf = self.get(f"{CONF_PLAYERS}/{player.player_id}") or {}
509        return SetupFlowContext(
510            kind="setup",
511            reason="user",
512            domain=player.provider.domain,
513            instance_id=player.provider.instance_id,
514            player_id=player.player_id,
515            setup_data=self._decrypt_values(raw_conf.get("setup_data") or {}),
516            values=self._decrypt_values(raw_conf.get("values") or {}),
517        )
518
519    def _protocol_children_with_setup_flow(
520        self, player: Player, *, needing_only: bool
521    ) -> list[Player]:
522        """
523        Return the player's protocol child players that implement a setup flow.
524
525        Covers the wrapper case: a universal player, or a native player wrapping
526        protocol children, whose own setup is a no-op but whose linked protocol
527        outputs still require pairing/credentials.
528
529        :param player: The (wrapper) player whose protocol children to inspect.
530        :param needing_only: Only return children that currently need setup.
531        """
532        children: list[Player] = []
533        seen: set[str] = set()
534        for output_protocol in player.output_protocols:
535            child_id = output_protocol.output_protocol_id
536            if output_protocol.is_native or child_id in seen:
537                continue
538            seen.add(child_id)
539            child = self.mass.players.get_player(child_id)
540            if child is None or not child.implements_setup_flow:
541                continue
542            if needing_only and not child.needs_setup:
543                continue
544            children.append(child)
545        return children
546
547    async def _run_child_selection_flow(
548        self, children: list[Player], session: SetupSession
549    ) -> None:
550        """
551        Run the wrapper flow that lets the user pick which protocol child to set up.
552
553        The selection form is owned by the parent; once a child is picked the session is
554        re-pointed at that child so its flow's steps localize and its ``finish()`` persists
555        under the child's own config.
556        """
557        options = [
558            ConfigValueOption(
559                value=child.player_id, title=f"{child.display_name} ({child.provider.name})"
560            )
561            for child in children
562        ]
563        values = await session.form(
564            [
565                ConfigEntry(
566                    key="child",
567                    type=ConfigEntryType.STRING,
568                    required=True,
569                    default_value=children[0].player_id,
570                    options=options,
571                )
572            ],
573            step_id="select_child",
574        )
575        child_id = str(values["child"])
576        child = next((candidate for candidate in children if candidate.player_id == child_id), None)
577        if child is None:
578            raise AbortFlow("nothing_to_configure")
579        child_raw = self.get(f"{CONF_PLAYERS}/{child_id}") or {}
580        session.retarget(
581            domain=child.provider.domain,
582            instance_id=child.provider.instance_id,
583            player_id=child_id,
584            setup_data=self._decrypt_values(child_raw.get("setup_data") or {}),
585            values=self._decrypt_values(child_raw.get("values") or {}),
586        )
587        await child.run_setup_flow(session)
588
589    async def _get_setup_flow_module(self, manifest: ProviderManifest) -> Any | None:
590        """
591        Import (lazily) the provider's setup_flow module, or None when it has none.
592
593        A provider without a setup_flow module needs no setup input at all.
594        """
595        # ensure the provider's requirements are installed and its package imports
596        # cleanly first: the setup_flow submodule may rely on those requirements
597        await load_provider_module(manifest.domain, manifest.requirements)
598        module_path = f"music_assistant.providers.{manifest.domain}.setup_flow"
599        try:
600            return await import_module_in_thread(module_path)
601        except ModuleNotFoundError as err:
602            if err.name == module_path:
603                # the provider ships no setup_flow module: it needs no setup input
604                return None
605            # an import *inside* setup_flow.py failed: an actual bug, surface it
606            raise
607
608    @property
609    def _setup_flows(self) -> dict[str, ActiveSetupFlow]:
610        """Return the registry of running flows (created lazily)."""
611        if self._flows is None:
612            self._flows = {}
613        return self._flows
614
615    def _get_flow(self, flow_id: str) -> ActiveSetupFlow:
616        """Return the running flow for the given id."""
617        if flow := self._setup_flows.get(flow_id):
618            return flow
619        msg = f"Unknown (or finished) setup flow: {flow_id}"
620        raise KeyError(msg)
621
622    def _check_flow_permission(self, flow: ActiveSetupFlow) -> None:
623        """Verify the calling user holds the scope the flow's start command required."""
624        # imported here: the webserver helpers pull in the full auth stack,
625        # which must not be imported with the config controller at startup
626        from music_assistant.controllers.webserver.helpers.auth_middleware import (  # noqa: PLC0415
627            get_current_user,
628            has_scope,
629        )
630
631        user = get_current_user()
632        # no user context means an internal (server-side) caller, which is trusted
633        if user is not None and not has_scope(user, flow.required_scope):
634            raise InsufficientPermissions(
635                f"This action requires the {flow.required_scope.value} scope"
636            )
637
638    def _synthesized_step(
639        self,
640        step_type: FlowStepType,
641        translation_owner: str,
642        *,
643        result: dict[str, str] | None = None,
644        reason: str | None = None,
645    ) -> SetupFlowStep:
646        """Return a terminal step for a flow that ended before a session was needed."""
647        return SetupFlowStep(
648            flow_id=uuid4().hex,
649            step_id="finish" if step_type == FlowStepType.FINISH else "abort",
650            type=step_type,
651            result=result,
652            reason=reason,
653            translation_owner=translation_owner,
654        )
655
656    def _reconfigure_reason(self, last_error: Any) -> FlowReason:
657        """Derive the reconfigure flow reason from the provider's stored last_error."""
658        if not last_error:
659            return "user"
660        # legacy settings may still hold a plain string last_error
661        error_code = last_error.get("error_code") if isinstance(last_error, dict) else None
662        if error_code in _AUTH_ERROR_CODES:
663            return "auth"
664        return "error"
665
666    def _encrypt_values(self, values: dict[str, ConfigValueType]) -> dict[str, Any]:
667        """Return a copy of the values with all string values encrypted (at-rest form)."""
668        return {
669            key: self.encrypt_string(value) if isinstance(value, str) else value
670            for key, value in values.items()
671        }
672
673    def _decrypt_values(self, values: dict[str, Any]) -> dict[str, Any]:
674        """Return a copy of the values with all (encrypted) string values decrypted."""
675        return {
676            key: self.decrypt_string(value) if isinstance(value, str) else value
677            for key, value in values.items()
678        }
679
680    def _schedule_flow_sweep(self) -> None:
681        """Arm the periodic idle-flow sweeper (idempotent)."""
682        if self._flow_sweep_handle is not None:
683            return
684        self._flow_sweep_handle = self.mass.loop.call_later(
685            FLOW_SWEEP_INTERVAL, self._sweep_idle_flows
686        )
687
688    def _sweep_idle_flows(self) -> None:
689        """Abort flows that have been idle for longer than the TTL."""
690        self._flow_sweep_handle = None
691        if self.mass.closing:
692            return
693        now = time.monotonic()
694        for flow in list(self._setup_flows.values()):
695            current_step = flow.session.current_step
696            if (
697                current_step is not None
698                and current_step.expires_at is not None
699                and current_step.expires_at > time.time()
700            ):
701                # the step advertises a (longer) countdown to the user; the step
702                # deadline machinery guarantees the flow terminates on its own
703                continue
704            if now - flow.session.last_activity >= IDLE_FLOW_TTL:
705                self.mass.create_task(self._abort_flow(flow, "timed_out"))
706        if self._setup_flows:
707            self._schedule_flow_sweep()
708