app_music_servers.py

# AudioMuse-AI - https://github.com/NeptuneHub/AudioMuse-AI
# Copyright (C) 2025 NeptuneHub
# SPDX-License-Identifier: AGPL-3.0-only
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License v3.0. See the LICENSE file
# in the project root or <https://github.com/NeptuneHub/AudioMuse-AI/blob/main/LICENSE>

"""REST API for the media-server registry (multi-server support).

Lets the setup wizard and the shared server dropdown list, add, edit, test,
delete and set-default the configured media servers, and trigger the
cross-server matching sweep. Every configured server is always active; there
is no per-server enable/disable state. Listing is available to any
authenticated user (credentials masked); every mutation is admin-only,
mirroring the setup page - EXCEPT during first-run setup, where the wizard
itself is the (unauthenticated) caller and /api/setup is already open.

Main Features:
* CRUD over the registry with masked secrets and a preserve-on-mask update.
* Connection testing and per-server catalogue-matching sweep enqueue.
* Usable by the first-run setup wizard, so a fresh install can configure its
  media servers here before any admin account exists.
"""

import logging
import uuid

from flask import Blueprint, g, jsonify, request
from rq.job import Job

import config
import rq_job_state
from app_helper import coerce_db_details, redis_conn, rq_queue_high, save_task_status, send_stop_job_command
from database import get_db, missing_required_creds, get_active_main_task
from app_server_context import (
    merge_creds,
    server_public_dict,
    servers_for_ui,
)
from tasks import provider_probe
from tasks.mediaserver import registry

logger = logging.getLogger(__name__)

music_servers_bp = Blueprint('music_servers_bp', __name__)

_SUPPORTED_TYPES = ('jellyfin', 'emby', 'navidrome', 'lyrion', 'plex', 'ampache')


def _setup_in_progress():
    """True while the first-run setup wizard is the caller.

    Set by the auth barrier when the install still needs setup: no admin
    account exists yet, so there is nobody to authenticate as, and the whole
    /api/setup surface (which writes these same credentials) is already open in
    that window. It closes the moment setup completes, after which every
    mutation here is admin-only again.
    """
    return bool(getattr(g, 'setup_needed', False))


def _is_admin_caller():
    return (
        _setup_in_progress()
        or (not config.AUTH_ENABLED)
        or getattr(g, 'auth_role', None) == 'admin'
    )


def _forbid_non_admin():
    if _is_admin_caller():
        return None
    return jsonify({"error": "Forbidden"}), 403


def _validate_type(server_type):
    return isinstance(server_type, str) and server_type.lower() in _SUPPORTED_TYPES


def _as_bool(value):
    """Parse a JSON flag. A non-UI caller may send the STRING "false", which is
    truthy in Python - and would silently promote its server to default."""
    if isinstance(value, str):
        return value.strip().lower() in ('true', '1', 'yes', 'on')
    return bool(value)


def _apply_default_to_config():
    """Propagate a default-server change to every process.

    The registry row that just changed IS the source of truth; the config module
    globals are only its projection. Reload them here for this process and
    request a restart so workers re-import config and re-project the row. No
    values are written anywhere - the registry was already updated by the caller.
    """
    import restart_manager

    config.refresh_config()
    restart_manager.publish_restart_request()


def _cancel_active_sweeps():
    """Revoke queued/running alignment sweeps so a consolidated one replaces them.

    Surgical per-sweep cancel: touches ONLY the stale sweep jobs, never the RQ
    queues or other task_status rows, so a running analysis (or any other job)
    keeps going when a server is added or edited. The REVOKED row is written
    first so the sweep's cooperative cancellation check picks it up even when
    the RQ commands fail.
    """
    cancelled = []
    try:
        db = get_db()
        cur = db.cursor()
        try:
            cur.execute(
                "SELECT task_id FROM task_status WHERE task_type = 'server_sweep' "
                "AND status NOT IN (%s, %s, %s)",
                (config.TASK_STATUS_SUCCESS, config.TASK_STATUS_FAILURE,
                 config.TASK_STATUS_REVOKED),
            )
            rows = cur.fetchall()
        finally:
            cur.close()
    except Exception:
        logger.exception("Could not look up active sweeps to supersede")
        return cancelled
    for row in rows:
        stale_task_id = row[0]
        try:
            save_task_status(
                stale_task_id, 'server_sweep', config.TASK_STATUS_REVOKED,
                progress=100,
                details={'message': 'Superseded by a new alignment covering all servers.'},
            )
            cancelled.append(stale_task_id)
        except Exception:
            logger.exception("Could not revoke superseded sweep %s", stale_task_id)
            continue
        try:
            job = Job.fetch(stale_task_id, connection=redis_conn)
            status = job.get_status(refresh=False)
            if rq_job_state.is_running_status(status):
                send_stop_job_command(redis_conn, stale_task_id)
            elif rq_job_state.is_alive_status(status):
                job.cancel()
        except Exception:
            logger.exception("RQ cleanup failed for superseded sweep %s", stale_task_id)
    return cancelled


def _enqueue_sweep(at_front=False):
    """Replace any queued/running sweep with one alignment of every server.

    Adding several servers back to back cancels the previous alignment each time
    and starts a fresh one, so the newest sweep always covers every not-yet-aligned
    server and no stale sweep for an outdated server set keeps running.

    Refuses while a cleaning run is live: both prune track_server_map against a
    catalogue snapshot taken minutes earlier, so an overlap lets one delete the
    mappings the other just wrote.
    """
    active = get_active_main_task(task_type='cleaning')
    if active:
        logger.warning(
            "Server alignment not enqueued: cleaning task %s is still %s. "
            "Re-run the alignment once it finishes.",
            active['task_id'], active['status'],
        )
        return None

    superseded = _cancel_active_sweeps()
    task_id = str(uuid.uuid4())
    try:
        save_task_status(
            task_id, 'server_sweep', config.TASK_STATUS_PENDING,
            details={
                'message': 'Server alignment queued for all servers.',
                'full_refresh': True,
            },
        )
        rq_queue_high.enqueue(
            'tasks.multiserver_sync.sweep_all_secondary_servers',
            kwargs={'task_id': task_id},
            job_id=task_id,
            job_timeout=-1,
            at_front=at_front,
        )
        if superseded:
            logger.info(
                "Superseded %d active sweep(s) with consolidated alignment %s",
                len(superseded), task_id,
            )
        return task_id
    except Exception:
        logger.exception("Failed to enqueue the server alignment")
        return None


def _latest_sweep_task():
    try:
        db = get_db()
        cur = db.cursor()
        try:
            cur.execute(
                "SELECT task_id, status, progress, details FROM task_status "
                "WHERE task_type = 'server_sweep' ORDER BY timestamp DESC LIMIT 1"
            )
            row = cur.fetchone()
        finally:
            cur.close()
        if not row:
            return None
        details = coerce_db_details(row[3])
        if not isinstance(details, dict):
            details = {}
        message = details.get('status_message') or details.get('message') or ''
        return {'task_id': row[0], 'status': row[1], 'progress': row[2] or 0, 'message': message}
    except Exception:
        logger.exception("Could not load latest sweep task")
        return None


def _name_taken(name, exclude_server_id=None):
    wanted = (name or '').strip()
    if not wanted:
        return False
    server = registry.get_server_by_name(wanted)
    return server is not None and server['server_id'] != exclude_server_id


def _missing_cred_keys(server_type, creds):
    """Required-but-empty cred keys for ``server_type`` (url/token/... style keys)."""
    return missing_required_creds(server_type, creds)


def _placeholder_default():
    """The default server row when it is only init_db's credential-less seed.

    A fresh install always carries one (seeded from an unconfigured config), and
    it is not a server anybody can reach: the first real server added has to
    take its place, or setup could never complete. Returns None when the default
    is a properly configured server (or when there is no default at all, which
    the registry already resolves by making the new server the default).
    """
    try:
        default = registry.get_default_server()
    except Exception:
        logger.exception("Could not read the default server")
        return None
    if default is None:
        return None
    if _missing_cred_keys(default['server_type'], default['creds']):
        return default
    return None


def _drop_unused_placeholder(placeholder):
    """Delete the seed row once a real server has replaced it as the default.

    Kept when it owns track mappings: that would mean a once-working server
    whose credentials were cleared, and its catalogue bindings are not ours to
    throw away - it just stays as a secondary for the admin to fix or remove.
    """
    try:
        if registry.mapped_count(placeholder['server_id']):
            return False
        registry.delete_server(placeholder['server_id'])
        logger.info(
            "Removed the unconfigured seed server '%s'; '%s' is the default now.",
            placeholder['name'], registry.get_default_server_id(),
        )
        return True
    except Exception:
        logger.exception("Could not remove the unconfigured seed server")
        return False


@music_servers_bp.route('/api/servers', methods=['GET'])
def list_servers():
    """List configured media servers plus the default id.

    Admins receive each server's masked credentials (to prefill the setup editor);
    non-admins receive only the fields the menu dropdown needs, with no creds.
    """
    payload = servers_for_ui()
    payload['sweep_task'] = _latest_sweep_task()
    if not _is_admin_caller():
        for server in payload['servers']:
            server.pop('creds', None)
    return jsonify(payload)


@music_servers_bp.route('/api/servers', methods=['POST'])
def add_server():
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    data = request.get_json(silent=True) or {}
    name = (data.get('name') or '').strip()
    server_type = (data.get('server_type') or '').strip().lower()
    creds = data.get('creds') or {}
    if not name:
        return jsonify({"error": "Server name is required."}), 400
    if not _validate_type(server_type):
        return jsonify({"error": f"server_type must be one of {list(_SUPPORTED_TYPES)}."}), 400
    if not isinstance(creds, dict):
        return jsonify({"error": "creds must be an object."}), 400
    if _name_taken(name):
        return jsonify({"error": f"A server named '{name}' already exists; names must be unique."}), 400
    make_default = _as_bool(data.get('make_default', False))
    missing = _missing_cred_keys(server_type, creds)
    if missing:
        return jsonify(
            {"error": f"Missing required credentials for {server_type}: {', '.join(missing)}."}
        ), 400
    placeholder = _placeholder_default()
    if placeholder is not None and not make_default:
        logger.info(
            "No usable default server is configured; '%s' becomes the default.", name
        )
        make_default = True
    server_id = registry.add_server(
        name=name,
        server_type=server_type,
        creds=creds,
        music_libraries=data.get('music_libraries') or '',
        make_default=make_default,
    )
    sweep_task_id = None
    created = registry.get_server(server_id)
    if created and created['is_default']:
        if placeholder is not None and placeholder['server_id'] != server_id:
            _drop_unused_placeholder(placeholder)
        _apply_default_to_config()
    sweep_task_id = _enqueue_sweep()
    body = server_public_dict(created)
    body['sweep_task_id'] = sweep_task_id
    return jsonify(body), 201


@music_servers_bp.route('/api/servers/<server_id>', methods=['PUT', 'PATCH'])
def update_server(server_id):
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    existing = registry.get_server(server_id)
    if existing is None:
        return jsonify({"error": "Unknown server."}), 404
    data = request.get_json(silent=True) or {}
    server_type = data.get('server_type')
    if server_type is not None:
        server_type = server_type.strip().lower()
        if not _validate_type(server_type):
            return jsonify({"error": f"server_type must be one of {list(_SUPPORTED_TYPES)}."}), 400
    new_name = data.get('name').strip() if isinstance(data.get('name'), str) else None
    if isinstance(data.get('name'), str) and not new_name:
        return jsonify({"error": "Server name cannot be empty"}), 400
    if new_name and _name_taken(new_name, exclude_server_id=server_id):
        return jsonify({"error": f"A server named '{new_name}' already exists; names must be unique."}), 400
    creds = None
    if 'creds' in data and isinstance(data['creds'], dict):
        creds = merge_creds(existing['creds'], data['creds'])
    is_default = registry.get_default_server_id(get_db()) == server_id
    # The DEFAULT server is validated too: it is the one config projects onto
    # every unbound provider call, so saving it credential-less breaks the whole
    # install (and the providers would silently fall back to stale config values).
    if server_type is not None or creds is not None:
        effective_type = server_type or existing['server_type']
        effective_creds = creds if creds is not None else existing['creds']
        missing = _missing_cred_keys(effective_type, effective_creds)
        if missing:
            return jsonify(
                {"error": f"Missing required credentials for {effective_type}: {', '.join(missing)}."}
            ), 400
    registry.update_server(
        server_id,
        name=new_name,
        server_type=server_type,
        creds=creds,
        music_libraries=data.get('music_libraries'),
    )
    sweep_task_id = None
    if is_default:
        _apply_default_to_config()
    # Sweep only on changes that can alter track matching; renames never
    # re-match the catalogue.
    new_libraries = data.get('music_libraries')
    needs_sweep = (
        (server_type is not None and server_type != existing['server_type'])
        or (creds is not None and creds != (existing['creds'] or {}))
        or (new_libraries is not None and new_libraries != existing['music_libraries'])
    )
    if needs_sweep:
        sweep_task_id = _enqueue_sweep()
    body = server_public_dict(registry.get_server(server_id))
    body['sweep_task_id'] = sweep_task_id
    return jsonify(body)


@music_servers_bp.route('/api/servers/<server_id>', methods=['DELETE'])
def delete_server(server_id):
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    try:
        deleted = registry.delete_server(server_id)
    except ValueError as exc:
        return jsonify({"error": str(exc)}), 400
    if not deleted:
        return jsonify({"error": "Unknown server."}), 404
    return jsonify({"deleted": server_id})


@music_servers_bp.route('/api/servers/<server_id>/default', methods=['POST'])
def set_default_server(server_id):
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    if registry.get_server(server_id) is None:
        return jsonify({"error": "Unknown server."}), 404
    registry.set_default(server_id)
    sweep_task_id = _enqueue_sweep()
    _apply_default_to_config()
    payload = servers_for_ui()
    payload['sweep_task_id'] = sweep_task_id
    return jsonify(payload)


def _parse_probe_request():
    forbidden = _forbid_non_admin()
    if forbidden:
        return None, None, forbidden
    data = request.get_json(silent=True) or {}
    server_type = (data.get('server_type') or '').strip().lower()
    creds = data.get('creds') or {}
    if not _validate_type(server_type):
        error = jsonify({"error": f"server_type must be one of {list(_SUPPORTED_TYPES)}."}), 400
        return None, None, error
    server_id = data.get('server_id')
    if server_id:
        existing = registry.get_server(server_id)
        if existing is not None:
            creds = merge_creds(existing['creds'], creds)
    return server_type, creds, None


@music_servers_bp.route('/api/servers/test', methods=['POST'])
def test_server():
    server_type, creds, error = _parse_probe_request()
    if error is not None:
        return error
    try:
        result = provider_probe.test_connection(server_type, creds)
    except Exception:
        logger.exception("Media server test connection failed")
        return jsonify({"ok": False, "error": "Connection test failed; check container logs."}), 200
    return jsonify(result)


@music_servers_bp.route('/api/servers/libraries', methods=['POST'])
def server_libraries():
    server_type, creds, error = _parse_probe_request()
    if error is not None:
        return error
    try:
        return jsonify(provider_probe.list_libraries(server_type, creds))
    except Exception:
        logger.exception("Media server list libraries failed")
        return jsonify({"libraries": [], "unsupported": True}), 200


@music_servers_bp.route('/api/servers/align', methods=['POST'])
def align_servers():
    """Align every secondary server against the default (no-op when aligned)."""
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    task_id = _enqueue_sweep(at_front=True)
    if task_id is None:
        return jsonify({"error": "Could not enqueue the alignment; check container logs."}), 500
    return jsonify({"enqueued": True, "task_id": task_id}), 202


@music_servers_bp.route('/api/servers/<server_id>/sweep', methods=['POST'])
def sweep_server(server_id):
    forbidden = _forbid_non_admin()
    if forbidden:
        return forbidden
    if registry.get_server(server_id) is None:
        return jsonify({"error": "Unknown server."}), 404
    task_id = str(uuid.uuid4())
    try:
        save_task_status(
            task_id, 'server_sweep', config.TASK_STATUS_PENDING,
            details={
                'message': 'Server matching sweep queued.',
                'full_refresh': True,
            },
        )
        rq_queue_high.enqueue(
            'tasks.multiserver_sync.sweep_server',
            args=(server_id,),
            kwargs={'task_id': task_id},
            job_id=task_id,
            job_timeout=-1,
        )
    except Exception:
        logger.exception("Failed to enqueue matching sweep for server %s", server_id)
        return jsonify({"error": "Could not enqueue the sweep; check container logs."}), 500
    return jsonify({"enqueued": True, "task_id": task_id, "job_id": task_id, "server_id": server_id}), 202