bambuddy/backend/app/api/routes/settings.py

import io
import json
import logging
import os
import zipfile
from datetime import datetime
from pathlib import Path

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.auth import (
    RequestPrinterScope,
    RequireAdminIfAuthEnabled,
    RequirePermissionIfAuthEnabled,
    ScopedCaller,
    caller_is_api_key,
    is_auth_enabled,
    require_auth_if_enabled,
    require_energy_cost_update,
)
from backend.app.core.config import APP_VERSION, settings as app_settings
from backend.app.core.database import get_db
from backend.app.core.permissions import Permission
from backend.app.core.printer_scope import PrinterScope
from backend.app.models.settings import Settings
from backend.app.models.user import User
from backend.app.schemas.settings import ADMIN_ONLY_SETTING_KEYS, AppSettings, AppSettingsUpdate
from backend.app.services import camera_light
from backend.app.services.secret_fields import SETTINGS_SECRETS, SecretConfigError, mask_fields, merge_partial

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/settings", tags=["settings"])

DEFAULT_SETTINGS = AppSettings()

# Sensitive credential fields blanked for API-key callers. Everyone else sees
# them as the placeholder (write-only secrets, see services/secret_fields).
_SENSITIVE_FIELDS_FOR_API_KEY = (*SETTINGS_SECRETS, "octoeverywhere_api_key")


async def get_setting(db: AsyncSession, key: str) -> str | None:
    """Get a single setting value by key."""
    result = await db.execute(select(Settings).where(Settings.key == key))
    setting = result.scalar_one_or_none()
    return setting.value if setting else None


# Accepted spellings for a boolean settings value. Settings live in a VARCHAR
# column and every reader compares them as strings, so these are normalised to
# "true"/"false" on the way in. The sets are deliberately generous: these
# endpoints are part of the documented REST surface, reached by scripts and by
# Home Assistant rest_command, where "True", "1" and "on" are all natural.
_TRUTHY_SETTING_VALUES = frozenset({"true", "1", "yes", "on"})
_FALSY_SETTING_VALUES = frozenset({"false", "0", "no", "off"})


def setting_is_true(value: object) -> bool:
    """Return True if a *stored* settings value means "on".

    Deliberately narrower than the spellings ``normalize_bool_setting`` accepts:
    it matches only what every other reader in the codebase treats as on
    (``value.lower() == "true"``). Submitted values are canonicalised on write,
    so a stored value is always "true"/"false"/""; accepting "1" or "on" here
    would make this function disagree with the rest of the app about any legacy
    row containing them.

    A bool is tolerated for the case of a row written before values were
    normalised, where SQLite coerced a raw bool into the VARCHAR column.
    """
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    return str(value).strip().lower() == "true"


def normalize_bool_setting(key: str, value: object) -> str:
    """Coerce a boolean-ish settings value to the canonical "true"/"false".

    Raises HTTPException(400) for values with no sensible interpretation, so an
    API client gets a message naming the field instead of a 500.

    A JSON boolean is the natural thing for an API client to send, and before
    this normalisation it caused two distinct failures on
    ``PUT /settings/spoolman``: ``bool.lower()`` raised AttributeError, and the
    raw bool was written into a VARCHAR column, which SQLite silently coerces
    to 1/0 while asyncpg rejects outright. Both surfaced as an opaque 500.
    """
    if isinstance(value, bool):  # must precede the int branch — bool is an int
        return "true" if value else "false"
    if isinstance(value, int):
        if value in (0, 1):
            return "true" if value else "false"
        raise HTTPException(400, f"{key} must be a boolean; got the number {value}")
    if isinstance(value, str):
        candidate = value.strip().lower()
        if not candidate:
            # Empty is stored verbatim rather than normalised to "false".
            # get_spoolman_settings reads these with ``or "<default>"``, so an
            # empty stored value means "use the default" — and two of them
            # (spoolman_report_partial_usage, auto_add_unknown_rfid) default to
            # ON. Rewriting "" to "false" would silently switch them off for any
            # client that submits a blank value.
            return ""
        if candidate in _TRUTHY_SETTING_VALUES:
            return "true"
        if candidate in _FALSY_SETTING_VALUES:
            return "false"
        raise HTTPException(400, f"{key} must be a boolean; got {value!r}")
    raise HTTPException(400, f"{key} must be a boolean; got {type(value).__name__}")


def normalize_str_setting(key: str, value: object) -> str:
    """Return a string settings value, rejecting types that would store garbage.

    ``str()`` on a dict or list would persist its repr, so those are refused
    rather than silently written. Numbers are accepted and stringified: a port
    or a bare host submitted unquoted is a plausible client mistake, not a
    reason to fail the request.
    """
    if isinstance(value, str):
        return value
    if value is None:
        return ""
    if isinstance(value, bool | int | float):
        return str(value)
    raise HTTPException(400, f"{key} must be a string; got {type(value).__name__}")


async def get_external_base_url(db: AsyncSession) -> str:
    """Base URL for links Bambuddy hands to the outside world (no trailing slash).

    ``external_url`` is optional and has no default, so anything that must be
    absolute — a login link in an e-mail, the one-tap outcome verdict links
    (#1898), whose Telegram/ntfy buttons are dropped for a relative URL — falls
    back to APP_URL and finally to the dev origin.
    """
    import os

    external_url = await get_setting(db, "external_url")
    if external_url:
        return external_url.rstrip("/")
    return os.environ.get("APP_URL", "http://localhost:5173").rstrip("/")


async def get_external_login_url(db: AsyncSession) -> str:
    """Get the external URL for the login page.

    Uses external_url from settings if available, otherwise falls back to APP_URL env var.

    Args:
        db: Database session

    Returns:
        Full URL to the login page
    """
    return await get_external_base_url(db) + "/login"


async def set_setting(db: AsyncSession, key: str, value: str) -> None:
    """Set a single setting value."""
    from backend.app.core.db_dialect import upsert_setting

    await upsert_setting(db, Settings, key, value)


# Settings stored as booleans / numbers. Storage is a VARCHAR column, so
# _build_settings_response() parses these back, and update_settings() refuses
# an explicit null for them: a null is stored as the literal "None", which
# reads back as False for a boolean and is not a number at all.
_BOOL_SETTING_KEYS = frozenset(
    {
        "auto_archive",
        "save_thumbnails",
        "capture_finish_photo",
        "finish_photo_restore_plate",
        "spoolman_enabled",
        "spoolman_disable_weight_sync",
        "spoolman_report_partial_usage",
        "auto_add_unknown_rfid",
        "disable_filament_warnings",
        "prefer_lowest_filament",
        "check_updates",
        "check_printer_firmware",
        "include_beta_updates",
        "announcements_enabled",
        "announcements_all_users",
        "ftp_retry_enabled",
        "mqtt_enabled",
        "mqtt_use_tls",
        "ha_enabled",
        "per_printer_mapping_expanded",
        "prometheus_enabled",
        "user_notifications_enabled",
        "queue_drying_enabled",
        "queue_drying_block",
        "ambient_drying_enabled",
        "print_drying_enabled",
        "require_plate_clear",
        "queue_shortest_first",
        # default_bed_levelling / default_flow_cali / default_nozzle_offset_cali
        # are tri-state strings (off/on/auto) — parsed via the raw-string else
        # branch; the TriState validator coerces legacy "true"/"false" rows.
        "default_vibration_cali",
        "default_layer_inspect",
        "default_timelapse",
        "default_confirm_outcome",
        "confirm_outcome_external_prints",
        "confirm_default_good_on_plate_clear",
        "billing_enabled",
        "printer_kill_switch_enabled",
        "ldap_enabled",
        "ldap_auto_provision",
        "local_login_enabled",
        "preheat_enabled",
        "queue_keep_bed_warm",
    }
)

_FLOAT_SETTING_KEYS = frozenset(
    {
        "default_filament_cost",
        "energy_cost_per_kwh",
        "camera_light_delay",
        "ams_temp_good",
        "ams_temp_fair",
        "library_disk_warning_gb",
        "low_stock_threshold",
    }
)

# String settings limited to a fixed set of values, which a stored "None"
# would break when the settings response is built.
_ENUM_SETTING_KEYS = frozenset({"energy_price_source"})

_INT_SETTING_KEYS = frozenset(
    {
        "ams_humidity_good",
        "ams_humidity_fair",
        "ams_history_retention_days",
        "printer_sensor_history_retention_days",
        "ftp_retry_count",
        "ftp_retry_delay",
        "ftp_timeout",
        "mqtt_port",
        "stagger_group_size",
        "stagger_interval_minutes",
        "forecast_global_lead_time_days",
        "location_sensor_poll_interval",
        "finance_budget_reset_day",
        "session_max_hours",
        "pipeline_max_copies",
        "preheat_max_wait_seconds",
        "preheat_soak_seconds",
        "queue_keep_warm_bed_temp",
        "queue_keep_warm_max_minutes",
        "queue_max_concurrent_uploads",
        "ambient_drying_sustained_minutes",
        "pause_warning_minutes",
        "octoeverywhere_poll_interval",
    }
)


async def _build_settings_response(db: AsyncSession, is_api_key: bool = False, mask: bool = True) -> AppSettings:
    """Build the full settings response.

    Secrets are blank for API-key callers and the placeholder for everyone else;
    ``mask=False`` is for server-side use only (comparing with an update).
    """
    settings_dict = DEFAULT_SETTINGS.model_dump()

    result = await db.execute(select(Settings))
    for setting in result.scalars().all():
        if setting.key not in settings_dict:
            continue
        if setting.key in _BOOL_SETTING_KEYS:
            settings_dict[setting.key] = setting.value.lower() == "true"
        elif setting.key in _FLOAT_SETTING_KEYS or setting.key in _INT_SETTING_KEYS:
            # A value that does not parse (the literal "None" from an old
            # null save, or a hand-edited row) keeps the default instead of
            # taking the whole settings response down with it.
            parse = int if setting.key in _INT_SETTING_KEYS else float
            try:
                settings_dict[setting.key] = parse(setting.value)
            except (TypeError, ValueError):
                logger.warning("Setting %s has an unparseable value; using the default", setting.key)
        elif setting.key in [
            # Nullable floats. Settings storage stringifies None to the literal
            # "None", which must read back as null here -- not as the default
            # the _FLOAT_SETTING_KEYS branch above falls back to (#2905).
            "ams_temp_alarm",
        ]:
            try:
                settings_dict[setting.key] = float(setting.value)
            except (TypeError, ValueError):
                settings_dict[setting.key] = None
        elif setting.key == "default_printer_id":
            settings_dict[setting.key] = int(setting.value) if setting.value and setting.value != "None" else None
        elif setting.key == "open_in_slicer":
            # None means "inherit from preferred_slicer" (#1329). The PUT path
            # serializes None as the literal string "None"; strip it back so
            # the frontend sees a true null and falls back as intended.
            settings_dict[setting.key] = setting.value if setting.value and setting.value != "None" else None
        else:
            settings_dict[setting.key] = setting.value

    ha_settings = await get_homeassistant_settings(db)
    settings_dict.update(ha_settings)

    # ldap_bind_password is never returned to any caller (mask=False is
    # server-side only)
    if mask or is_api_key:
        settings_dict["ldap_bind_password"] = ""

    # The saved key is write-only, including for authenticated administrators.
    settings_dict["octoeverywhere_api_key_configured"] = bool(settings_dict["octoeverywhere_api_key"].strip())
    settings_dict["octoeverywhere_api_key"] = ""

    if is_api_key:
        for field in _SENSITIVE_FIELDS_FOR_API_KEY:
            if field in settings_dict:
                settings_dict[field] = ""
    elif mask:
        settings_dict = mask_fields(settings_dict, SETTINGS_SECRETS)

    return AppSettings(**settings_dict)


@router.get("", response_model=AppSettings)
@router.get("/", response_model=AppSettings)
async def get_settings(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
    _is_api_key: bool = Depends(caller_is_api_key),
):
    """Get all application settings."""
    return await _build_settings_response(db, is_api_key=_is_api_key)


def _as_stored_setting(value: object) -> str:
    """The text the settings writer below stores for ``value``."""
    if isinstance(value, bool):
        return "true" if value else "false"
    if value is None:
        return "None"
    return str(value)


def _same_setting_value(new: object, old: object) -> bool:
    """Whether saving ``new`` stores exactly what ``old`` is stored as."""
    return _as_stored_setting(new) == _as_stored_setting(old)


@router.put("/", response_model=AppSettings)
async def update_settings(
    settings_update: AppSettingsUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
):
    """Update application settings."""
    update_data = settings_update.model_dump(exclude_unset=True)

    # Write-only secrets: the placeholder keeps the stored value, and a kept
    # secret does not follow a changed destination.
    stored = (await _build_settings_response(db, mask=False)).model_dump()
    secrets = dict(SETTINGS_SECRETS)
    if stored.get("ha_token_from_env"):
        # The deployment's own token, not a stored secret to enter again: the
        # URL may change (the placeholder sent back still writes nothing).
        secrets["ha_token"] = ()
    try:
        update_data = merge_partial(stored, update_data, secrets)
    except SecretConfigError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from None

    # Omitted/null preserves the saved key; an explicit empty string removes it.
    if update_data.get("octoeverywhere_api_key") is None:
        update_data.pop("octoeverywhere_api_key", None)

    # Only one failure detection provider may control a print at a time.
    if update_data.get("obico_enabled") and update_data.get("octoeverywhere_enabled"):
        raise HTTPException(status_code=400, detail="Enable only one AI failure detection provider at a time.")
    if update_data.get("octoeverywhere_enabled"):
        update_data["obico_enabled"] = False
    elif update_data.get("obico_enabled"):
        update_data["octoeverywhere_enabled"] = False

    # An explicit null for a boolean or numeric setting has no meaning -- these
    # are not clearable -- and would be stored as the literal "None".
    null_keys = sorted(
        key
        for key, value in update_data.items()
        if value is None and key in (_BOOL_SETTING_KEYS | _FLOAT_SETTING_KEYS | _INT_SETTING_KEYS | _ENUM_SETTING_KEYS)
    )
    if null_keys:
        raise HTTPException(status_code=422, detail=f"These settings cannot be null: {', '.join(null_keys)}")

    # Settings that decide who can sign in are admin-only, like the user routes.
    # Only a real change counts: the settings page sends every field on each
    # save, so an unchanged value from a non-admin must pass.
    if current_user is not None and not current_user.is_admin:
        changed = sorted(
            key
            for key in update_data
            if key in ADMIN_ONLY_SETTING_KEYS and not _same_setting_value(update_data[key], stored.get(key))
        )
        if changed:
            raise HTTPException(
                status_code=403,
                detail=f"Only an administrator can change these settings: {', '.join(changed)}",
            )

    # Like an API key's or a provider's printers: a page still open from before
    # a printer delete would save the deleted id, which SQLite gives to the next
    # printer created (the validator has already checked it is a list of ints).
    # Only ids not stored yet are checked: the page sends the whole list, and an
    # id already there must not block every save.
    for key in ("obico_enabled_printers", "octoeverywhere_enabled_printers"):
        if not update_data.get(key):
            continue
        from backend.app.core.printer_scope import ensure_printers_exist

        try:
            stored_printers = json.loads(await get_setting(db, key) or "[]")
        except ValueError:
            stored_printers = []
        stored_printers = set(stored_printers) if isinstance(stored_printers, list) else set()
        added = [i for i in json.loads(update_data[key]) if i not in stored_printers]
        await ensure_printers_exist(db, added)

    # Safety refusals on disabling local login (#1589). Two failure modes
    # would otherwise lock everyone out of the install:
    #   1. No enabled OIDC provider exists — nobody could authenticate.
    #   2. The caller has no UserOIDCLink — they would lock themselves out
    #      even if other admins are linked.
    # Either case returns HTTP 400 instead of silently saving. The
    # ``BAMBUDDY_LOCAL_LOGIN=true`` env-var bypass on /auth/login is a
    # separate recovery path; the refusals here protect the *default*
    # configuration where the env var is absent.
    if update_data.get("local_login_enabled") is False:
        from backend.app.models.oidc_provider import OIDCProvider, UserOIDCLink

        enabled_count = await db.scalar(select(func.count(OIDCProvider.id)).where(OIDCProvider.is_enabled.is_(True)))
        if not enabled_count:
            raise HTTPException(
                status_code=400,
                detail="Cannot disable local login: no OIDC provider is enabled.",
            )
        if current_user is not None:
            caller_links = await db.scalar(
                select(func.count(UserOIDCLink.id)).where(UserOIDCLink.user_id == current_user.id)
            )
            if not caller_links:
                raise HTTPException(
                    status_code=400,
                    detail="Cannot disable local login: your account has no OIDC link, so you would lock yourself out.",
                )

    # Check if any MQTT settings are being updated
    mqtt_keys = {
        "mqtt_enabled",
        "mqtt_broker",
        "mqtt_port",
        "mqtt_username",
        "mqtt_password",
        "mqtt_topic_prefix",
        "mqtt_use_tls",
    }
    mqtt_updated = bool(mqtt_keys & set(update_data.keys()))

    # The page saves every field it shows on each change, so compare against
    # what is stored: only a real change of price source is worth a read.
    price_source_changed = False
    for key in ("energy_price_source", "energy_price_ha_entity"):
        if key in update_data and str(update_data[key]) != (await get_setting(db, key) or ""):
            price_source_changed = True

    for key, value in update_data.items():
        # Convert value to string for storage
        await set_setting(db, key, _as_stored_setting(value))

    await db.commit()
    # Expire all objects to ensure fresh reads after commit
    db.expire_all()

    if update_data.get("obico_enabled") is False:
        from backend.app.services.obico_detection import obico_detection_service

        await obico_detection_service.refresh_settings()
    if any(key.startswith("octoeverywhere_") for key in update_data):
        from backend.app.services.octoeverywhere_detection import octoeverywhere_detection_service

        await octoeverywhere_detection_service.refresh_settings()

    if {"camera_light_mode", "camera_light_delay"} & set(update_data.keys()):
        camera_light.invalidate_settings()

    if price_source_changed:
        # Read a newly chosen price sensor now, so the saved settings show its
        # price instead of the old one until the next hourly read (#1251).
        from backend.app.services.energy_price import current_price

        try:
            await current_price(db, remember=True)
            await db.commit()
        except Exception as e:
            logger.warning("Could not read the electricity price after saving its source: %s", e)

    # Reconfigure MQTT relay if any MQTT settings changed
    if mqtt_updated:
        try:
            from backend.app.services.mqtt_relay import mqtt_relay

            mqtt_settings = {
                "mqtt_enabled": (await get_setting(db, "mqtt_enabled") or "false") == "true",
                "mqtt_broker": await get_setting(db, "mqtt_broker") or "",
                "mqtt_port": int(await get_setting(db, "mqtt_port") or "1883"),
                "mqtt_username": await get_setting(db, "mqtt_username") or "",
                "mqtt_password": await get_setting(db, "mqtt_password") or "",
                "mqtt_topic_prefix": await get_setting(db, "mqtt_topic_prefix") or "bambuddy",
                "mqtt_use_tls": (await get_setting(db, "mqtt_use_tls") or "false") == "true",
            }
            await mqtt_relay.configure(mqtt_settings)
        except Exception:
            pass  # Don't fail the settings update if MQTT reconfiguration fails

    # Return updated settings, secrets as the placeholder
    return await _build_settings_response(db, is_api_key=False)


@router.patch("/", response_model=AppSettings)
@router.patch("", response_model=AppSettings)
async def patch_settings(
    settings_update: AppSettingsUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
):
    """Partially update application settings (same as PUT, for REST compatibility)."""
    return await update_settings(settings_update, db, _)


class ElectricityPriceUpdate(BaseModel):
    """Payload for ``POST /settings/electricity-price`` (#1356).

    Mirrors the field name documented in ``wiki/features/energy.md`` so the
    Home Assistant ``rest_command`` example needs only a URL change, not a
    payload change. Plain non-negative float; tariffs can go as low as 0.0 in
    some markets (e.g. free hours).
    """

    energy_cost_per_kwh: float = Field(ge=0)


@router.post("/electricity-price", response_model=AppSettings)
async def update_electricity_price(
    payload: ElectricityPriceUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: ScopedCaller = Depends(require_energy_cost_update()),
    _is_api_key: bool = Depends(caller_is_api_key),
):
    """Update the per-kWh electricity cost used by the energy-tracking pipeline.

    This is the only settings field writable via API key, gated by the
    ``can_update_energy_cost`` toggle on the key. JWT users still need the
    standard ``SETTINGS_UPDATE`` permission. See #1356 for the rationale —
    the general ``PATCH /settings`` route remains denied for API keys because
    it can rewrite SMTP/LDAP/MQTT credentials, which is a much wider surface
    than the documented dynamic-tariff use case requires.
    """
    await set_setting(db, "energy_cost_per_kwh", str(payload.energy_cost_per_kwh))
    await db.commit()
    db.expire_all()
    return await _build_settings_response(db, is_api_key=_is_api_key)


@router.post("/reset", response_model=AppSettings)
async def reset_settings(
    db: AsyncSession = Depends(get_db, scope="function"),
    _admin: User | None = RequireAdminIfAuthEnabled(),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
):
    """Reset all settings to defaults."""
    # Delete all settings
    result = await db.execute(select(Settings))
    for setting in result.scalars().all():
        await db.delete(setting)

    await db.commit()
    camera_light.invalidate_settings()

    return DEFAULT_SETTINGS


@router.get("/default-sidebar-order")
async def get_default_sidebar_order(
    db: AsyncSession = Depends(get_db, scope="function"),
):
    """Get the admin-set default sidebar order.

    Intentionally unauthenticated: non-admin users need to read this value to apply
    the default sidebar order, but may lack SETTINGS_READ permission.
    The value is non-sensitive (sidebar item IDs only).
    """
    value = await get_setting(db, "default_sidebar_order")
    return {"default_sidebar_order": value or ""}


# Fields exposed via /ui-preferences without SETTINGS_READ. Each entry MUST be
# non-sensitive (no credentials, no PII, no secret tokens) — granting SETTINGS_READ
# also grants visibility of SMTP/LDAP/MQTT passwords and similar, so the goal of
# this endpoint is exactly to NOT require that permission for UI rendering hints.
# When adding a field here, confirm it doesn't carry anything sensitive.
_UI_PREFERENCE_FIELDS: tuple[str, ...] = (
    "require_plate_clear",
    "check_printer_firmware",
    "camera_view_mode",
    "time_format",
    "date_format",
    "drying_presets",
    "ams_humidity_thresholds",
    "ams_humidity_good",
    "ams_humidity_fair",
    "ams_temp_good",
    "ams_temp_fair",
    # ams_temp_alarm is deliberately NOT here. This endpoint is unauthenticated
    # and exists so the UI can colour readings without SETTINGS_READ; the good /
    # fair bands are what the printer card colours by. The alarm threshold
    # changes no rendering anywhere -- only SettingsPage reads it, and that is
    # behind the settings permissions already (#2905).
    "bed_cooled_threshold",
    # Temperature / fan-speed presets for the printer-card popovers. Numbers
    # only; no PII / credentials.
    "nozzle_temp_presets",
    "bed_temp_presets",
    "chamber_temp_presets",
    "fan_speed_presets",
)


@router.get("/ui-preferences")
async def get_ui_preferences(db: AsyncSession = Depends(get_db, scope="function")):
    """Get the curated subset of settings that any page needs to render correctly.

    Intentionally not gated on SETTINGS_READ — every authenticated user (and
    every page that loads for them) needs these fields, but granting SETTINGS_READ
    would also grant visibility of secrets (SMTP/LDAP/MQTT credentials, etc.).
    Same pattern as /default-sidebar-order (#1293).

    Reuses _build_settings_response so the typed values match what /settings
    returns for fields with the same name — bool/int/float/str types stay in
    sync without a separate type-coercion path.
    """
    full = await _build_settings_response(db, is_api_key=False)
    dumped = full.model_dump()
    return {key: dumped[key] for key in _UI_PREFERENCE_FIELDS if key in dumped}


# Install configuration the app shell reads before it can render correctly.
#
# Deliberately a second list rather than more entries in _UI_PREFERENCE_FIELDS.
# That one is served to anyone at all, on the recorded grounds that its contents
# are "public defaults that ship with the app" (test_route_auth_coverage.py), and
# its field set is pinned by a test written to make anyone adding to it stop and
# think. These fields are not defaults -- they are facts about how this
# particular deployment is configured -- so they get their own endpoint at their
# own trust level instead of stretching that charter to fit them.
_UI_FLAG_FIELDS: tuple[str, ...] = (
    # The sidebar hides Finance unless billing is on. Layout read this from
    # GET /settings, which requires SETTINGS_READ, so for a non-admin the query
    # 403'd, the value arrived undefined, `undefined !== true` held, and the
    # entry was hidden from exactly the users cost_centers:read_own exists to
    # serve. The page itself was reachable by URL the whole time (#3023).
    "billing_enabled",
    # Same 403, opposite outcome. That gate tests `=== false`, which undefined
    # never satisfies, so an administrator who turned user notifications off
    # still left the entry showing -- to precisely the non-admins it governs.
    "user_notifications_enabled",
    # Not gates, but read by the shell and equally undefined for a non-admin:
    # the sponsor prompt fell back to EUR whatever the install uses, and the
    # update check ran even where it had been switched off.
    "currency",
    "check_updates",
    # Read by the print dialog, queue, file manager, inventory and SpoolBuddy
    # pages, which users without SETTINGS_READ open too: the 403 dropped them
    # all to their defaults (with billing on, the print dialog hid the cost
    # centre picker and the print was then refused for lacking one).
    "queue_shortest_first",
    "use_slicer_api",
    "open_in_slicer",
    "preferred_slicer",
    "disable_filament_warnings",
    "prefer_lowest_filament",
    "per_printer_mapping_expanded",
    "default_filament_cost",
    "default_bed_levelling",
    "default_flow_cali",
    "default_vibration_cali",
    "default_layer_inspect",
    "default_timelapse",
    "default_nozzle_offset_cali",
    "default_confirm_outcome",
    "stagger_group_size",
    "stagger_interval_minutes",
    "library_disk_warning_gb",
    "low_stock_threshold",
    "location_sensor_poll_interval",
    "location_sensor_alert_defaults",
    "pipeline_max_copies",
    "language",
    "dark_style",
    "dark_background",
    "dark_accent",
    "light_style",
    "light_background",
    "light_accent",
    # Which inventory a new spool goes to (unknown-tag prompt), and the
    # forecast's default lead time.
    "spoolman_enabled",
    "forecast_global_lead_time_days",
    # Preselected in the print dialog; withheld below from a caller who may
    # not use that printer.
    "default_printer_id",
)


@router.get("/ui-flags")
async def get_ui_flags(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = Depends(require_auth_if_enabled),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Install configuration the app shell needs, for any signed-in user.

    Gated on being authenticated rather than on ``SETTINGS_READ``. The sidebar
    has to know whether billing is enabled before it can decide whether to offer
    Finance, and ``SETTINGS_READ`` cannot be the price of knowing that -- it also
    grants sight of the SMTP, LDAP and MQTT credentials.

    ``require_auth_if_enabled`` returns ``None`` when auth is switched off
    entirely, which is the case /ui-preferences was left ungated for. That is the
    distinction the two endpoints draw: "works when there is no auth" is not the
    same statement as "readable by anyone", and conflating them is what put a
    settings read in front of a permission that was never meant to require one.
    """
    full = await _build_settings_response(db, is_api_key=False)
    dumped = full.model_dump()
    flags = {key: dumped[key] for key in _UI_FLAG_FIELDS if key in dumped}
    if flags.get("default_printer_id") is not None and not printer_scope.allows(flags["default_printer_id"]):
        flags["default_printer_id"] = None
    # Whether G-code snippets are set up, not the snippets themselves.
    flags["has_gcode_snippets"] = bool(dumped.get("gcode_snippets"))
    return flags


@router.get("/check-ffmpeg")
async def check_ffmpeg(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Check if ffmpeg is installed and available.

    Gated on ``SETTINGS_READ`` (audit finding I4 — the binary path was
    leaking the host filesystem layout to unauthenticated callers).
    ``require_permission_if_auth_enabled`` returns ``None`` only when
    auth is disabled (in which case there's no privacy boundary to
    enforce); otherwise it raises 401/403 before we get here.
    """
    from backend.app.services.camera import get_ffmpeg_path

    ffmpeg_path = get_ffmpeg_path()
    return {
        "installed": ffmpeg_path is not None,
        "path": ffmpeg_path,
    }


@router.get("/spoolman")
async def get_spoolman_settings(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Get Spoolman integration settings."""
    spoolman_enabled = await get_setting(db, "spoolman_enabled") or "false"
    spoolman_url = await get_setting(db, "spoolman_url") or ""
    spoolman_sync_mode = await get_setting(db, "spoolman_sync_mode") or "auto"
    spoolman_disable_weight_sync = await get_setting(db, "spoolman_disable_weight_sync") or "false"
    spoolman_report_partial_usage = await get_setting(db, "spoolman_report_partial_usage") or "true"
    auto_add_unknown_rfid = await get_setting(db, "auto_add_unknown_rfid") or "true"

    return {
        "spoolman_enabled": spoolman_enabled,
        "spoolman_url": spoolman_url,
        "spoolman_sync_mode": spoolman_sync_mode,
        "spoolman_disable_weight_sync": spoolman_disable_weight_sync,
        "spoolman_report_partial_usage": spoolman_report_partial_usage,
        "auto_add_unknown_rfid": auto_add_unknown_rfid,
    }


@router.put("/spoolman")
async def update_spoolman_settings(
    settings: dict,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
):
    """Update Spoolman integration settings.

    The body is a free-form dict rather than a schema, so each value is
    normalised before it is persisted — see ``normalize_bool_setting`` for why
    a JSON boolean used to produce a 500 here.
    """
    if "spoolman_enabled" in settings:
        was_enabled = setting_is_true(await get_setting(db, "spoolman_enabled"))
        new_val = normalize_bool_setting("spoolman_enabled", settings["spoolman_enabled"])
        now_enabled = new_val == "true"
        await set_setting(db, "spoolman_enabled", new_val)

        # Nothing is deleted on a mode change (#2812). Each mode keeps its slot
        # assignments in its own table, so both can hold rows at once and the
        # toggle is reversible: switching to Spoolman to see what it does, then
        # switching back, returns you to the assignments you had.
        #
        # This used to empty the other mode's table on every toggle. The reason
        # was real -- checks that read both tables would let a row in the mode
        # you are not using answer for the mode you are -- but the cost was that
        # inspecting a mode destroyed your configuration, with no confirmation
        # and no way back, and the deletion was unfiltered across every printer.
        # The readers that could be confused now ask which mode is active
        # (``spoolman_owns_assignments``), which is where that decision belongs:
        # the mode is a property of the install, not of the rows.
        if was_enabled != now_enabled:
            logger.info(
                "Inventory mode switched to %s; slot assignments in both tables kept",
                "Spoolman" if now_enabled else "built-in",
            )
    if "spoolman_url" in settings:
        await set_setting(db, "spoolman_url", normalize_str_setting("spoolman_url", settings["spoolman_url"]))
    if "spoolman_sync_mode" in settings:
        await set_setting(
            db, "spoolman_sync_mode", normalize_str_setting("spoolman_sync_mode", settings["spoolman_sync_mode"])
        )
    for bool_key in ("spoolman_disable_weight_sync", "spoolman_report_partial_usage", "auto_add_unknown_rfid"):
        if bool_key in settings:
            await set_setting(db, bool_key, normalize_bool_setting(bool_key, settings[bool_key]))

    spoolman_changed = "spoolman_enabled" in settings or "spoolman_url" in settings

    await db.commit()
    db.expire_all()

    if spoolman_changed:
        from backend.app.services.location_service import maybe_sync_spoolman_locations

        if await maybe_sync_spoolman_locations(db):
            await db.commit()

    # Return updated settings
    return await get_spoolman_settings(db)


async def get_homeassistant_settings(db: AsyncSession) -> dict:
    """
    Get Home Assistant integration settings.
    Environment variables (HA_URL, HA_TOKEN) take precedence over database settings.
    """
    import os

    # Check environment variables first
    ha_url_env = os.environ.get("HA_URL")
    ha_token_env = os.environ.get("HA_TOKEN")

    # Fall back to database values
    ha_url = ha_url_env or await get_setting(db, "ha_url") or ""
    ha_token = ha_token_env or await get_setting(db, "ha_token") or ""
    ha_enabled_db = await get_setting(db, "ha_enabled") or "false"

    # Track which settings come from environment
    ha_url_from_env = bool(ha_url_env)
    ha_token_from_env = bool(ha_token_env)
    ha_env_managed = ha_url_from_env and ha_token_from_env

    # Auto-enable when both env vars are set, otherwise use database value
    if ha_url_env and ha_token_env:
        ha_enabled = True
    else:
        ha_enabled = ha_enabled_db.lower() == "true"

    return {
        "ha_enabled": ha_enabled,
        "ha_url": ha_url,
        "ha_token": ha_token,
        "ha_url_from_env": ha_url_from_env,
        "ha_token_from_env": ha_token_from_env,
        "ha_env_managed": ha_env_managed,
    }


async def create_backup_zip(output_path: Path | None = None) -> tuple[Path, str]:
    """Create a complete backup ZIP (database + all data directories).

    If output_path is given, the ZIP is written there.
    Otherwise a temporary file is created (caller must clean up).
    Returns (zip_path, filename).
    """
    import shutil
    import tempfile

    from backend.app.core.db_dialect import is_sqlite

    base_dir = app_settings.base_dir
    filename = f"bambuddy-backup-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"

    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)

        if is_sqlite():
            from sqlalchemy import text

            from backend.app.core.database import engine

            db_path = Path(app_settings.database_url.replace("sqlite+aiosqlite:///", ""))

            # Checkpoint WAL to ensure all data is in main db file
            async with engine.begin() as conn:
                await conn.execute(text("PRAGMA wal_checkpoint(TRUNCATE)"))

            # Copy database file
            shutil.copy2(db_path, temp_path / "bambuddy.db")
        else:
            # PostgreSQL: export to a portable SQLite file via SQLAlchemy.
            # This makes backups restorable on both SQLite and Postgres installs.
            import json
            import sqlite3

            from sqlalchemy import create_engine as create_sync_engine

            from backend.app.core.database import Base, engine

            backup_db_path = temp_path / "bambuddy.db"
            metadata = Base.metadata

            # Build the portable SQLite schema with SQLAlchemy's own DDL rather
            # than a hand-rolled CREATE TABLE. metadata.create_all() emits the
            # exact schema a native SQLite install gets — NOT NULL, DEFAULT
            # (server_default=func.now() → CURRENT_TIMESTAMP), foreign keys,
            # unique constraints and indexes. The previous name+type-only
            # rebuild dropped all of these, so a Postgres→SQLite restore left
            # server_default columns (e.g. spoolbuddy_devices.created_at) with
            # no DEFAULT — SQLAlchemy omits such columns on INSERT and the DB
            # then wrote NULL, which 500'd on the next read (#2526). Using the
            # real DDL also keeps the #1333 BLOB guard: LargeBinary still
            # renders as BLOB, so OIDC icon bytes survive the round trip.
            schema_engine = create_sync_engine(f"sqlite:///{backup_db_path}")
            try:
                metadata.create_all(schema_engine)
            finally:
                schema_engine.dispose()

            dst = sqlite3.connect(str(backup_db_path))

            # Export data from Postgres to SQLite
            async with engine.connect() as conn:
                for table in metadata.sorted_tables:
                    result = await conn.execute(table.select())
                    rows = result.fetchall()
                    if not rows:
                        continue
                    columns = list(result.keys())
                    placeholders = ", ".join(["?"] * len(columns))
                    col_list = ", ".join(columns)
                    insert_sql = f"INSERT INTO {table.name} ({col_list}) VALUES ({placeholders})"  # noqa: S608  # nosec B608 — table/column names from ORM metadata, not user input

                    def _serialize_row(row):
                        return tuple(json.dumps(v) if isinstance(v, (list, dict)) else v for v in row)

                    dst.executemany(insert_sql, [_serialize_row(row) for row in rows])

            dst.commit()
            dst.close()
            logger.info("PostgreSQL backup exported to portable SQLite format")

        # Copy data directories (if they exist)
        dirs_to_backup = [
            ("archive", base_dir / "archive"),
            ("virtual_printer", base_dir / "virtual_printer"),
            ("plate_calibration", app_settings.plate_calibration_dir),
            ("icons", base_dir / "icons"),
            ("projects", base_dir / "projects"),
            ("overlay-branding", base_dir / "overlay-branding"),
        ]

        for name, src_dir in dirs_to_backup:
            if src_dir.exists() and any(src_dir.iterdir()):
                try:
                    shutil.copytree(
                        src_dir, temp_path / name
                    )  # SEC-PATH-OK: name iterates the dirs_to_backup tuple of constant strings ("archive", "virtual_printer", ...)
                except shutil.Error as e:
                    logger.warning("Some files in %s could not be copied: %s", name, e)
                except PermissionError as e:
                    logger.warning("Permission denied copying %s: %s", name, e)

        # Say which version made this, so a restore that cannot import it can
        # name the versions rather than a list of columns. Backups from before
        # this existed simply have no manifest, and restore treats the version
        # as unknown.
        import json as _json

        manifest = {
            "format": 1,
            "app_version": APP_VERSION,
            "created_at": datetime.now().isoformat(timespec="seconds"),
            "database": "sqlite" if is_sqlite() else "postgresql",
        }
        (temp_path / "manifest.json").write_text(_json.dumps(manifest, indent=2) + "\n")

        # Include the MFA encryption key as a ZIP top-level entry alongside
        # bambuddy.db. Without it, encrypted client_secret / TOTP secret rows
        # would be unrecoverable after restore on a host without MFA_ENCRYPTION_KEY set.
        from backend.app.core.paths import resolve_data_dir

        mfa_key_src = resolve_data_dir() / ".mfa_encryption_key"
        if mfa_key_src.exists() and mfa_key_src.is_file():
            try:
                shutil.copy2(mfa_key_src, temp_path / ".mfa_encryption_key")
            except OSError as exc:
                logger.error(
                    "Could not include MFA encryption key in backup (%s). "
                    "The backup ZIP will not contain the key — restore on a "
                    "keyless host will fail for encrypted secrets.",
                    exc,
                )
                raise

        # Create ZIP
        if output_path is not None:
            zip_file = (
                output_path / filename
            )  # SEC-PATH-OK: filename = f"bambuddy-backup-{datetime.now()...}.zip" generated in create_backup_zip itself
        else:
            fd, tmp = tempfile.mkstemp(suffix=".zip")
            os.close(fd)
            zip_file = Path(tmp)

        with zipfile.ZipFile(zip_file, "w", zipfile.ZIP_DEFLATED) as zf:
            for file_path in temp_path.rglob("*"):
                if file_path.is_file():
                    arcname = file_path.relative_to(temp_path)
                    zf.write(file_path, arcname)

    return zip_file, filename


@router.get("/backup")
async def create_backup(
    db: AsyncSession = Depends(get_db, scope="function"),
    _admin: User | None = RequireAdminIfAuthEnabled(),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_BACKUP),
):
    """Create a complete backup (database + all files) as a ZIP download."""
    from starlette.background import BackgroundTask

    try:
        zip_file, filename = await create_backup_zip()
        return FileResponse(
            path=zip_file,
            filename=filename,
            media_type="application/zip",
            background=BackgroundTask(lambda: zip_file.unlink(missing_ok=True)),
        )
    except Exception as e:
        logger.error("Backup failed: %s", e, exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"success": False, "message": "Backup failed. Check server logs for details."},
        )


class BackupSchemaIncompatible(Exception):
    """The backup has no value for a column this version requires.

    A backup carries the schema of the install that made it. Restoring it into
    a different version means the destination can have NOT NULL columns the
    backup never heard of -- either because that version is older and still has
    a column since removed (``user_wallets.currency``, dropped in #3123), or
    because it is newer and has added one. Most such columns have a default and
    can simply be filled. The ones that cannot are what this reports, and it has
    to be reported BEFORE the restore drops anything: the Postgres import wipes
    every table in the first transaction, so a failure halfway leaves the
    install with an empty schema and the previous data gone.
    """


def _missing_required_columns(pg_table, src_columns: set[str]):
    """Split the destination's NOT NULL columns that the backup lacks.

    Returns ``(injectable, db_filled, unfillable)``:

    * ``injectable`` -- ``{name: value}`` from the model's Python-side default.
      These are invisible to the import's raw SQL: SQLAlchemy applies a
      ``default=`` on ORM and Core inserts, never on ``text()``, and
      ``create_all`` emits no DDL default for one. So a column like
      ``currency VARCHAR(3) NOT NULL`` with ``default="EUR"`` arrives with
      nothing to put in it unless we put it there.
    * ``db_filled`` -- has a server default or is the autoincrement key; the
      database fills it when the column is left out of the INSERT.
    * ``unfillable`` -- nothing can supply a value. The backup is incompatible.
    """
    injectable: dict = {}
    db_filled: list[str] = []
    unfillable: list[str] = []

    for col in pg_table.columns:
        if col.nullable or col.name in src_columns:
            continue
        if col.default is not None:
            arg = col.default.arg
            injectable[col.name] = arg(None) if callable(arg) else arg
        elif col.server_default is not None or col.primary_key:
            db_filled.append(col.name)
        else:
            unfillable.append(col.name)

    return injectable, db_filled, unfillable


def check_backup_schema_compatible(sqlite_path: Path, backup_version: str | None = None) -> None:
    """Raise if this version cannot import that backup. Touches nothing.

    Only the cross-engine path needs this. A SQLite install restores by copying
    the backup's pages, schema included, and `init_db()` migrates it forward
    afterwards; the Postgres import instead recreates the schema from THIS
    process's ORM and then inserts the backup's columns into it.
    """
    import sqlite3

    from backend.app.core.database import Base

    src = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
    try:
        src_tables = {
            row[0]
            for row in src.execute(
                "SELECT name FROM sqlite_master WHERE type='table' "
                "AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'archive_fts%'"
            )
        }
        problems: list[str] = []
        # metadata.tables, not sorted_tables: the latter warns about the
        # library_files/library_folders/print_archives cycle, and nothing here
        # depends on the order.
        for name, pg_table in Base.metadata.tables.items():
            if name not in src_tables:
                continue
            # An empty table inserts nothing, so a column it cannot supply
            # cannot fail. Refusing a restore over one would be a false alarm.
            if src.execute(f'SELECT 1 FROM "{name}" LIMIT 1').fetchone() is None:  # noqa: S608  # nosec B608 — name comes from ORM metadata
                continue
            src_columns = {row[1] for row in src.execute(f'PRAGMA table_info("{name}")')}
            _, _, unfillable = _missing_required_columns(pg_table, src_columns)
            problems.extend(f"{name}.{col}" for col in unfillable)
    finally:
        src.close()

    if not problems:
        return

    made_by = f"The backup was made by Bambuddy {backup_version}, " if backup_version else "The backup "
    raise BackupSchemaIncompatible(
        "This backup cannot be restored by this version of Bambuddy. It carries no value for "
        f"{len(problems)} column(s) this version requires and cannot default: {', '.join(sorted(problems))}. "
        f"{made_by}and this install runs {APP_VERSION}. Restore it on the version that made it, or "
        "upgrade this install to that version. Nothing has been changed."
    )


def _read_backup_manifest(temp_path: Path) -> dict:
    """The backup's manifest.json, or {} for a backup made before it existed."""
    import json

    path = temp_path / "manifest.json"
    if not path.is_file():
        return {}
    try:
        data = json.loads(path.read_text())
    except (OSError, ValueError) as exc:
        logger.warning("Ignoring unreadable backup manifest: %s", exc)
        return {}
    return data if isinstance(data, dict) else {}


def _restored_value_fixes(table_name: str) -> dict[str, tuple[tuple[str, ...], str]]:
    """Columns of ``table_name`` whose backed-up value must be one of a set, with its fallback.

    Older versions stored any project priority or status; one longer than the
    column fails the whole import on PostgreSQL. The fallbacks are the ones
    project import and the start-up repair use.
    """
    if table_name != "projects":
        return {}
    from backend.app.models.project import PROJECT_PRIORITIES, PROJECT_STATUSES

    return {"priority": (PROJECT_PRIORITIES, "normal"), "status": (PROJECT_STATUSES, "active")}


async def _import_sqlite_to_postgres(sqlite_path: Path):
    """Import data from a SQLite database file into the current PostgreSQL database.

    Used for cross-database restore (SQLite backup → PostgreSQL).
    Reads all tables from the SQLite file and bulk-inserts into Postgres.
    """
    import sqlite3

    from sqlalchemy import text

    from backend.app.core.database import AUTOINCREMENT_TABLES, Base, _create_engine, highest_referenced_id

    # Before anything is dropped. The route checks this too, earlier and with
    # the backup's version in the message; this call is what makes the guarantee
    # a property of the import itself rather than of one caller.
    check_backup_schema_compatible(sqlite_path)

    # Create a temporary engine for the import (current engine was disposed)
    pg_engine = _create_engine()

    try:
        # Open SQLite file directly (sync — it's a local file read)
        src = sqlite3.connect(str(sqlite_path))
        src.row_factory = sqlite3.Row

        # Get list of tables from SQLite (skip internal/FTS tables)
        cursor = src.execute(
            "SELECT name FROM sqlite_master WHERE type='table' "
            "AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'archive_fts%'"
        )
        src_tables = {row["name"] for row in cursor.fetchall()}
        # Where SQLite's own counters stand: ids freed there are not given out
        # again on PostgreSQL either (a backup from before they were kept has none).
        src_counters: dict[str, int] = {}
        if src.execute("SELECT 1 FROM sqlite_master WHERE name = 'sqlite_sequence'").fetchone():
            src_counters = {row["name"]: row["seq"] for row in src.execute("SELECT name, seq FROM sqlite_sequence")}

        # Get Postgres tables from our ORM models
        metadata = Base.metadata
        pg_tables = set(metadata.tables.keys())

        # Only import tables that exist in both source and destination
        tables_to_import = src_tables & pg_tables
        sorted_tables = [t.name for t in metadata.sorted_tables if t.name in tables_to_import]

        # Phase 1: Drop all tables and recreate WITHOUT foreign keys.
        # This avoids all FK ordering/orphan issues during import; the
        # constraints go back on at the end, once every row has landed.
        async with pg_engine.begin() as conn:
            # Cap how long DROP TABLE will wait for AccessExclusiveLock so
            # any residual concurrent writer (per-printer MQTT clients
            # writing reactively, an AMS history recorder firing on its
            # hourly cadence) surfaces a fast `lock_timeout` error instead
            # of blocking the restore for 30 s or producing a deadlock.
            # SET LOCAL scopes to this transaction only; outside this
            # restore path the global default (no timeout) applies.
            await conn.execute(text("SET LOCAL lock_timeout = '10s'"))

            # Drop every existing table in the public schema with CASCADE
            # rather than `metadata.drop_all`. Two reasons:
            #   1. The user's live DB may carry orphan tables from removed
            #      features (e.g. the legacy `spoolman_slot_assignments`,
            #      `spoolman_k_profile`) that hold FK constraints back to
            #      ORM tables. `drop_all` doesn't know they exist and emits
            #      `DROP TABLE printers` without CASCADE — Postgres refuses
            #      and the whole restore aborts (#XXXX).
            #   2. Even within the metadata, `drop_all` is FK-ordered and
            #      breaks if a future schema rename leaves old constraints
            #      around. CASCADE is the right tool for a destructive
            #      restore: the user is intentionally wiping state.
            await conn.execute(
                text(
                    "DO $$ DECLARE r RECORD; BEGIN "
                    "FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public') LOOP "
                    "EXECUTE 'DROP TABLE IF EXISTS public.' || quote_ident(r.tablename) || ' CASCADE'; "
                    "END LOOP; END $$;"
                )
            )
            await conn.run_sync(metadata.create_all)

            # Now strip the foreign keys, at the database level.
            #
            # This used to be done by discarding each ForeignKeyConstraint
            # from `table.constraints` before `create_all`. That only
            # suppresses the inline REFERENCES clause inside CREATE TABLE:
            # `Table.foreign_key_constraints` is derived from the *columns'*
            # ForeignKey objects, which the discard never touched. When
            # `create_all` meets a dependency cycle it can't sort -- and
            # library_files / library_folders / print_archives are exactly
            # such a cycle -- it falls back to emitting those tables' keys
            # as separate ALTER TABLE ... ADD FOREIGN KEY statements read
            # straight from that property. Twelve constraints survived,
            # including library_files.folder_id, and because the same cycle
            # also drops the ordering edge from `sorted_tables` the child
            # table was imported before its parent and the restore died on
            # a ForeignKeyViolationError.
            #
            # Dropping them from pg_constraint instead is indifferent to how
            # create_all chose to emit them, so a future model cycle cannot
            # reintroduce this. It also keeps the app's global Base.metadata
            # untouched: the old code only put the constraints back *after*
            # the transaction, so a failure in here left the running process
            # with an FK-less metadata until restart.
            await conn.execute(
                text(
                    "DO $$ DECLARE r RECORD; BEGIN "
                    "FOR r IN (SELECT conrelid::regclass AS tbl, conname FROM pg_constraint "
                    "WHERE contype = 'f' AND connamespace = 'public'::regnamespace) LOOP "
                    "EXECUTE 'ALTER TABLE ' || r.tbl || ' DROP CONSTRAINT ' || quote_ident(r.conname); "
                    "END LOOP; END $$;"
                )
            )

        # Phase 2: Import data (no FKs to worry about)
        async with pg_engine.begin() as conn:
            # Import each table in dependency order (parents before children)
            for table_name in sorted_tables:
                rows = src.execute(f"SELECT * FROM {table_name}").fetchall()  # noqa: S608  # nosec B608
                if not rows:
                    continue

                # Filter to columns that exist in the Postgres table
                src_columns = rows[0].keys()
                pg_table = metadata.tables.get(table_name)
                pg_columns = {c.name for c in pg_table.columns} if pg_table is not None else set()
                columns = [c for c in src_columns if c in pg_columns]

                if not columns:
                    continue

                # Columns this schema requires that the backup does not have at
                # all. The block below handles a column PRESENT in the backup
                # with a NULL in it; one the backup never had is not in
                # `columns` and so never reached it -- which is how a backup
                # from an install without `user_wallets.currency` died on
                # NotNullViolationError against a version that still had it.
                injected, _db_filled, _unfillable = _missing_required_columns(pg_table, set(src_columns))
                if injected:
                    logger.info(
                        "Filling %s column(s) absent from the backup in %s: %s",
                        len(injected),
                        table_name,
                        ", ".join(sorted(injected)),
                    )

                insert_columns = columns + list(injected)
                col_list = ", ".join(insert_columns)
                param_list = ", ".join(f":{c}" for c in insert_columns)
                # ON CONFLICT DO NOTHING handles duplicate rows from SQLite (which doesn't enforce unique constraints)
                insert_sql = text(f"INSERT INTO {table_name} ({col_list}) VALUES ({param_list}) ON CONFLICT DO NOTHING")  # noqa: S608  # nosec B608

                # Identify columns that need type conversion (SQLite stores booleans
                # as int and datetimes as str — asyncpg requires native Python types)
                from datetime import datetime as dt, timezone

                bool_columns = set()
                datetime_columns = set()
                not_null_defaults = {}  # col_name -> default value for NOT NULL columns
                if pg_table is not None:
                    for col in pg_table.columns:
                        if col.name not in columns:
                            continue
                        col_type = str(col.type)
                        if col_type == "BOOLEAN":
                            bool_columns.add(col.name)
                        elif col_type in ("DATETIME", "TIMESTAMP WITHOUT TIME ZONE", "TIMESTAMP WITH TIME ZONE"):
                            datetime_columns.add(col.name)
                        # Track NOT NULL columns with defaults — older backups may have NULL
                        # for columns added after the backup was created
                        if not col.nullable:
                            if col.default is not None:
                                default = col.default.arg
                                if callable(default):
                                    default = default(None)
                                not_null_defaults[col.name] = default
                            elif col.server_default is not None:
                                # server_default=func.now() → use current timestamp
                                if col.name in datetime_columns:
                                    not_null_defaults[col.name] = "__now__"
                                else:
                                    # Try to extract literal server default
                                    sd = str(col.server_default.arg) if hasattr(col.server_default, "arg") else None
                                    if sd is not None:
                                        not_null_defaults[col.name] = sd

                # Naive UTC, as every stored timestamp is: the host's local
                # time put backfilled rows hours off (DB-C4).
                now = dt.now(timezone.utc).replace(tzinfo=None)
                value_fixes = _restored_value_fixes(table_name)

                def _convert_row(
                    row,
                    cols=columns,
                    bools=bool_columns,
                    dts=datetime_columns,
                    nn_defaults=not_null_defaults,
                    _now=now,
                    inject=injected,
                    fixes=value_fixes,
                ):
                    result = dict(inject)
                    for c in cols:
                        val = row[c]
                        if val is None and c in nn_defaults:
                            val = _now if nn_defaults[c] == "__now__" else nn_defaults[c]
                        if val is not None:
                            if c in bools:
                                val = bool(val)
                            elif c in dts and isinstance(val, str):
                                try:
                                    val = dt.fromisoformat(val)
                                except ValueError:
                                    pass
                        result[c] = val
                    for c, (allowed, fallback) in fixes.items():
                        if c in result and result[c] not in allowed:
                            result[c] = fallback
                    return result

                batch = [_convert_row(row) for row in rows]
                await conn.execute(insert_sql, batch)
                logger.info("Imported %d rows into %s", len(batch), table_name)

            # Reset sequences to max(id) + 1 for each table with an id column;
            # for AUTOINCREMENT_TABLES past SQLite's counter and every id still
            # referred to, so a deleted row's id is not given to a new one.
            for table_name in sorted_tables:
                try:
                    async with conn.begin_nested():
                        result = await conn.execute(text(f"SELECT MAX(id) FROM {table_name}"))  # noqa: S608  # nosec B608
                        max_id = result.scalar()
                        if table_name in AUTOINCREMENT_TABLES:
                            max_id = max(
                                max_id or 0,
                                src_counters.get(table_name) or 0,
                                await highest_referenced_id(conn, table_name),
                            )
                        if max_id:
                            seq_name = f"{table_name}_id_seq"
                            await conn.execute(text(f"SELECT setval('{seq_name}', {max_id})"))  # noqa: S608
                except Exception:
                    pass  # Table may not have an id column or sequence

        src.close()
        logger.info("Cross-database import complete: %d tables imported", len(tables_to_import))

        # Recreate FK constraints from ORM metadata, which Phase 1 left intact.
        # Use individual transactions so orphaned SQLite data doesn't block valid FKs.
        from sqlalchemy.schema import AddConstraint

        failed_fks = []
        for table in metadata.sorted_tables:
            for fk in table.foreign_key_constraints:
                try:
                    async with pg_engine.begin() as fk_conn:
                        await fk_conn.execute(AddConstraint(fk))
                except Exception as e:
                    # Name the constraint by what it links, not by `fk.name`:
                    # these are unnamed in the ORM, so that field is None and
                    # the warning used to read "print_archives.None" for every
                    # one of the five keys on that table -- unusable for
                    # working out which rows to go and look at.
                    cols = ", ".join(c.name for c in fk.columns)
                    target = fk.elements[0].target_fullname if fk.elements else "unknown"
                    failed_fks.append(f"{table.name}({cols}) -> {target}")
                    # Postgres puts the offending key in a DETAIL line; it
                    # names the exact orphan value, which is the one thing
                    # that turns this into an actionable report.
                    detail = next(
                        (ln.strip() for ln in str(e).splitlines() if ln.startswith("DETAIL:")),
                        str(e).splitlines()[0] if str(e) else e.__class__.__name__,
                    )
                    logger.info("FK %s(%s) -> %s not restored: %s", table.name, cols, target, detail)
        if failed_fks:
            logger.warning(
                "Could not restore %d FK constraints (orphaned data in the backup): %s. "
                "The data is restored and usable; those columns are simply no longer "
                "enforced. See the INFO lines above for the offending key in each case.",
                len(failed_fks),
                ", ".join(failed_fks),
            )

    finally:
        await pg_engine.dispose()


@router.post("/restore")
async def restore_backup(
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _admin: User | None = RequireAdminIfAuthEnabled(),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_RESTORE),
):
    """Restore from a complete backup ZIP.

    Replaces the database and all data directories from the backup ZIP.
    Requires a restart after restore.
    """
    import shutil
    import tempfile

    from fastapi import HTTPException

    from backend.app.core.database import close_all_connections, init_db, reinitialize_database
    from backend.app.core.db_dialect import is_sqlite
    from backend.app.services.virtual_printer import virtual_printer_manager

    base_dir = app_settings.base_dir

    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)

        # 1. Read and extract ZIP
        content = await file.read()

        # Check if it's a valid ZIP
        if not file.filename or not file.filename.endswith(".zip"):
            raise HTTPException(400, "Invalid backup file: must be a .zip file")

        try:
            with zipfile.ZipFile(io.BytesIO(content), "r") as zf:
                for name in zf.namelist():
                    # Reject path-traversal payloads: any entry whose resolved
                    # path escapes temp_path would allow writing arbitrary files
                    # on the host (ZipSlip / CVE-2006-5456).
                    dest = (
                        temp_path / name
                    ).resolve()  # SEC-PATH-OK: is_relative_to containment check below before extractall
                    # is_relative_to (Python 3.9+) covers both relative
                    # path-traversal (../etc/passwd) and absolute-path overrides
                    # (/etc/passwd) — str.startswith was vulnerable to
                    # prefix-collision attacks (e.g. /tmp/abc_evil/file passing
                    # a /tmp/abc prefix check).
                    if not dest.is_relative_to(temp_path.resolve()):
                        raise HTTPException(400, f"Invalid backup: unsafe path in ZIP: {name!r}")
                zf.extractall(temp_path)
        except zipfile.BadZipFile:
            raise HTTPException(400, "Invalid backup file: not a valid ZIP")

        # 2. Validate backup
        backup_db = temp_path / "bambuddy.db"
        if not backup_db.exists():
            raise HTTPException(400, "Invalid backup: missing bambuddy.db")

        # 2b. Can this version import this backup at all?
        #
        # Deliberately here: everything below has a side effect. The virtual
        # printer stops, background services stop, the MFA key file is
        # overwritten with the backup's -- and then the Postgres import drops
        # every table in its first transaction. A backup rejected at the INSERT
        # took the install's data with it and left the encrypted secrets under a
        # key that no longer matches. Nothing above this line has touched
        # anything.
        import sqlite3

        manifest = _read_backup_manifest(temp_path)
        backup_version = manifest.get("app_version")
        if backup_version:
            logger.info("Backup was created by Bambuddy %s; this install runs %s", backup_version, APP_VERSION)
        if not is_sqlite():
            try:
                check_backup_schema_compatible(backup_db, backup_version)
            except BackupSchemaIncompatible as exc:
                logger.error("Refusing backup: %s", exc)
                raise HTTPException(400, str(exc)) from exc
            except sqlite3.DatabaseError as exc:
                raise HTTPException(400, f"Invalid backup: bambuddy.db is not readable ({exc})") from exc

        try:
            import asyncio

            # 3. Stop the virtual printers if running (releases file locks).
            # stop_all, not the legacy configure(): that ignores its arguments
            # and re-syncs from the database not yet replaced, so their FTP,
            # MQTT and SSDP services ran on through the restore (vp-2). Like
            # the services below, they come back with the restart a
            # successful restore asks for.
            try:
                if virtual_printer_manager.is_enabled:
                    logger.info("Stopping virtual printers for restore...")
                    await virtual_printer_manager.stop_all()
                    await asyncio.sleep(1)
            except Exception as e:
                logger.warning("Failed to stop virtual printer: %s", e)

            # 3b. Pause timer-based background services BEFORE the DB swap.
            # close_all_connections() below only disposes the engine's pool,
            # not the asyncio tasks that opened sessions from it. The print
            # scheduler (30 s cadence), smart-plug snapshot loop (30 s), and
            # notification digest loop all
            # wake up and call async_session(), which lazily re-creates a
            # pool connection holding RowExclusiveLock on print_queue /
            # smart_plug_energy_snapshots / etc. The DROP TABLE CASCADE
            # pass in the PostgreSQL restore path needs AccessExclusiveLock
            # on every public table, producing an AB/BA deadlock and a
            # full restore rollback. Successful restore already requires a
            # container restart, so we don't restart the services here.
            try:
                from backend.app.services.notification_service import notification_service
                from backend.app.services.print_scheduler import scheduler as print_scheduler
                from backend.app.services.smart_plug_manager import smart_plug_manager

                logger.info("Pausing background services for restore...")
                print_scheduler.stop()
                smart_plug_manager.stop_scheduler()
                notification_service.stop_digest_scheduler()
                # In-flight loop iterations need a moment to commit + release
                # their DB sessions before we dispose() the engine pool.
                await asyncio.sleep(1.0)
            except Exception as e:
                logger.warning("Could not cleanly pause background services: %s", e)

            # 4. Close current database connections
            logger.info("Closing database connections...")
            await close_all_connections()

            # B1: Restore the MFA encryption key file BEFORE the database swap.
            # If the key write fails (OSError, RO disk, full disk, EACCES) we
            # can still abort while the live DB is intact. Doing this AFTER the
            # DB swap would leave the database with rows encrypted under the
            # backup's key but the running install holding only the old key —
            # every encrypted secret becomes unrecoverable.
            from backend.app.core.paths import resolve_data_dir

            mfa_key_src = temp_path / ".mfa_encryption_key"
            if mfa_key_src.exists() and mfa_key_src.is_file():
                dst_key = resolve_data_dir() / ".mfa_encryption_key"
                tmp_key = dst_key.parent / ".mfa_encryption_key.restore-tmp"
                try:
                    dst_key.parent.mkdir(parents=True, exist_ok=True)
                    # S1: atomic write with restrictive mode from creation.
                    # O_TRUNC because a stale tmp may exist from a prior
                    # failed restore attempt — we want to overwrite it.
                    fd = os.open(str(tmp_key), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
                    try:
                        os.write(fd, mfa_key_src.read_bytes())
                    finally:
                        os.close(fd)
                    # POSIX rename(2) — atomic when source/dest are on the
                    # same filesystem (we're staying inside dst_key.parent).
                    os.replace(str(tmp_key), str(dst_key))
                    # S9: warn if the FS doesn't enforce 0o600
                    actual_mode = dst_key.stat().st_mode & 0o777
                    if actual_mode != 0o600:
                        logger.warning(
                            "Restored MFA key file %s: filesystem did not enforce 0o600 "
                            "(actual: 0o%o). Key may be world-readable on Windows / SMB / FUSE.",
                            dst_key,
                            actual_mode,
                        )
                    logger.info("Restored .mfa_encryption_key from backup")
                except OSError as e:
                    logger.error(
                        "Could not write restored MFA key file to %s: %s — "
                        "aborting BEFORE database swap (DB unchanged).",
                        dst_key,
                        e,
                        exc_info=True,
                    )
                    raise HTTPException(
                        status_code=500,
                        detail=("Restore aborted: MFA key write failed. Database is unchanged. Check server logs."),
                    ) from e

            # 5. Replace database
            logger.info("Restoring database from backup...")
            if is_sqlite():
                db_path = Path(app_settings.database_url.replace("sqlite+aiosqlite:///", ""))
                # Use SQLite's online backup API instead of shutil.copy2.
                # The pragma at database.py:19 runs the live DB in WAL mode,
                # which means a naive file copy is unsafe: anything written
                # to the live DB before this call that hasn't been
                # checkpointed yet (seed_default_groups + init_db on first
                # start, plus whatever background heartbeats wrote during
                # the request window) sits in bambuddy.db-wal with valid
                # checksums. The route handler's own `db: Depends(get_db, scope="function")`
                # session also keeps a connection checked out across
                # engine.dispose(), holding fds to the WAL inode. With
                # `shutil.copy2` SQLite finds the stale WAL on the next
                # open and silently re-applies those page-level writes on
                # top of the restored DB, partially clobbering it with
                # fresh-install state — the user sees a "successful"
                # restore where most rows and settings have reverted to
                # defaults (#1211 / #668). The page-by-page backup API
                # opens both DBs as real SQLite connections, takes the
                # right locks, and routes new pages through the live DB's
                # own WAL — so concurrent open sessions see their own
                # snapshot until they close (transaction isolation) but
                # can't corrupt the restored state.
                import sqlite3

                src_conn = sqlite3.connect(str(backup_db))
                try:
                    dst_conn = sqlite3.connect(str(db_path))
                    try:
                        src_conn.backup(dst_conn)
                    finally:
                        dst_conn.close()
                finally:
                    src_conn.close()
            else:
                # Import SQLite backup into PostgreSQL
                logger.info("Importing SQLite backup into PostgreSQL...")
                await _import_sqlite_to_postgres(backup_db)

            # 6. Replace data directories
            # For Docker compatibility: clear contents then copy (don't delete mount points)
            dirs_to_restore = [
                ("archive", base_dir / "archive"),
                ("virtual_printer", base_dir / "virtual_printer"),
                ("plate_calibration", app_settings.plate_calibration_dir),
                ("icons", base_dir / "icons"),
                ("projects", base_dir / "projects"),
                ("overlay-branding", base_dir / "overlay-branding"),
            ]

            skipped_dirs = []
            for name, dest_dir in dirs_to_restore:
                src_dir = (
                    temp_path / name
                )  # SEC-PATH-OK: name iterates the dirs_to_restore tuple of constant strings ("archive", "virtual_printer", ...)
                if src_dir.exists():
                    logger.info("Restoring %s directory...", name)
                    try:
                        # Clear destination contents (not the dir itself - may be Docker mount)
                        if dest_dir.exists():
                            for item in dest_dir.iterdir():
                                try:
                                    if item.is_dir():
                                        shutil.rmtree(item)
                                    else:
                                        item.unlink()
                                except OSError as e:
                                    logger.warning("Could not delete %s: %s", item, e)
                        else:
                            dest_dir.mkdir(parents=True, exist_ok=True)
                        # Copy contents from backup
                        for item in src_dir.iterdir():
                            dest_item = dest_dir / item.name
                            if item.is_dir():
                                shutil.copytree(item, dest_item)
                            else:
                                shutil.copy2(item, dest_item)
                    except OSError as e:
                        logger.warning("Could not restore %s directory: %s", name, e)
                        skipped_dirs.append(name)

            # 7. Reset the encryption singleton so the migration that runs
            # inside init_db() picks up the restored key file (if a new one
            # was written above). Without this reset, _get_fernet would
            # return the cached Fernet instance built from the previous key.
            import backend.app.core.encryption as _enc_mod

            _enc_mod._fernet_instance = None
            _enc_mod._key_source = None
            _enc_mod._warn_shown = False

            # 8. Reinitialize the database engine and apply schema migrations so that
            # tables added after the backup was created (e.g. ams_labels) exist
            # immediately, without requiring a manual restart.
            await reinitialize_database()
            await init_db()

            # 9. Sign everyone out. The restored users and auth tokens come from
            # another point in time, possibly written by older code, so no
            # session issued before the restore should stay valid.
            from backend.app.core import database as _db_mod

            async with _db_mod.engine.begin() as conn:
                await _db_mod.invalidate_all_sessions(conn)
            # Open dashboards of signed-in users too: they are never signed in
            # again. API-key and anonymous ones are not closed outright (nothing
            # ended theirs), but the restored data decides what they may still see: the anonymous ones
            # go if the backup turns authentication on, and the refresh closes
            # those of a key the backup does not have (or has under another key).
            from backend.app.core.auth import invalidate_auth_enabled_cache
            from backend.app.core.websocket import ws_manager

            # Read from the restored data: this request has just cached the
            # setting it had before.
            invalidate_auth_enabled_cache()
            # Cloud presets were cached per user of the replaced user table.
            from backend.app.api.routes.cloud import forget_all_cloud_presets

            forget_all_cloud_presets()
            async with _db_mod.async_session() as auth_db:
                auth_on = await is_auth_enabled(auth_db)
            if auth_on:
                await ws_manager.close_signed_in()
                await ws_manager.close_anonymous()
            await ws_manager.refresh_printer_scopes()

            logger.info("Restore complete - restart required")
            message = "Backup restored successfully. Please restart Bambuddy for changes to take effect."
            if skipped_dirs:
                message += f" Note: Some directories could not be restored ({', '.join(skipped_dirs)})."
            return {
                "success": True,
                "message": message,
            }

        except HTTPException:
            # Preserve specific HTTP error responses raised inside the restore
            # body (e.g. the key-write OSError → 500). The blanket
            # except Exception below would otherwise swallow them and replace
            # the operator-facing detail with a generic message.
            raise
        except Exception as e:
            logger.error("Restore failed: %s", e, exc_info=True)
            return JSONResponse(
                status_code=500,
                content={"success": False, "message": "Restore failed. Check server logs for details."},
            )


@router.get("/network-interfaces")
async def get_network_interfaces(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Get available network interfaces with all IPs (primary + aliases)."""
    from backend.app.services.network_utils import get_all_interface_ips

    interfaces = get_all_interface_ips()
    return {"interfaces": interfaces}


@router.get("/virtual-printer/models")
async def get_virtual_printer_models(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Get available virtual printer models."""
    from backend.app.services.virtual_printer import (
        DEFAULT_VIRTUAL_PRINTER_MODEL,
        VIRTUAL_PRINTER_MODELS,
    )

    return {
        "models": VIRTUAL_PRINTER_MODELS,
        "default": DEFAULT_VIRTUAL_PRINTER_MODEL,
    }


def _legacy_vp_status(status: dict, printer_scope: PrinterScope) -> dict:
    """The first virtual printer's status, without what names its target for a limited caller."""
    if printer_scope.is_unrestricted:
        return status
    return {key: value for key, value in status.items() if key not in ("name", "serial", "target_printer_ip", "proxy")}


@router.get("/virtual-printer")
async def get_virtual_printer_settings(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get the settings this route still sets, and the virtual printers' status.

    Every other virtual printer setting lives on its own row
    (``GET /virtual-printers``).
    """
    from backend.app.services.virtual_printer import virtual_printer_manager

    tailscale_disabled_raw = await get_setting(db, "virtual_printer_tailscale_disabled")
    archive_name_source = await get_setting(db, "virtual_printer_archive_name_source")

    return {
        "tailscale_disabled": tailscale_disabled_raw == "true" if tailscale_disabled_raw else True,
        "archive_name_source": archive_name_source if archive_name_source in ("metadata", "filename") else "metadata",
        "status": _legacy_vp_status(virtual_printer_manager.get_status(), printer_scope),
    }


@router.put("/virtual-printer")
async def update_virtual_printer_settings(
    enabled: bool = None,
    access_code: str = None,
    mode: str = None,
    model: str = None,
    target_printer_id: int = None,
    remote_interface_ip: str = None,
    tailscale_disabled: bool = None,
    archive_name_source: str = None,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Legacy single-printer settings: the archive name source and the first
    virtual printer's Tailscale switch.

    Every other virtual printer setting lives on its own row, set with
    ``PUT /virtual-printers/{id}``; the rest of this route's parameters are
    refused (vp-3).
    """
    from sqlalchemy import select

    from backend.app.models.virtual_printer import VirtualPrinter as FirstVP

    current_target_id_str = await get_setting(db, "virtual_printer_target_printer_id")
    current_target_id = int(current_target_id_str) if current_target_id_str else None
    # One bridging to a printer the caller can't use is not theirs to change,
    # and they can't aim it at such a printer.
    # The first virtual printer is the one these settings reach.
    first_vp = (await db.execute(select(FirstVP).order_by(FirstVP.position, FirstVP.id).limit(1))).scalar_one_or_none()
    if not printer_scope.allows(current_target_id) or (
        first_vp is not None and not printer_scope.allows(first_vp.target_printer_id)
    ):
        return JSONResponse(status_code=404, content={"detail": "Virtual printer not found"})
    if target_printer_id is not None and not printer_scope.allows(target_printer_id):
        return JSONResponse(status_code=400, content={"detail": f"Printer with ID {target_printer_id} not found"})
    # Each virtual printer keeps these itself now; this route wrote them to
    # settings keys nothing reads and answered 200 as if applied (vp-3). The
    # stored legacy values are not checked either: an old install's stale
    # mode or deleted target printer refused the settings this still serves.
    unread = [
        name
        for name, value in (
            ("enabled", enabled),
            ("access_code", access_code),
            ("mode", mode),
            ("model", model),
            ("target_printer_id", target_printer_id),
            ("remote_interface_ip", remote_interface_ip),
        )
        if value is not None
    ]
    if unread:
        return JSONResponse(
            status_code=400,
            content={
                "detail": f"{', '.join(unread)} cannot be set here any more: set them for each virtual printer "
                "with PUT /api/v1/virtual-printers/{id}"
            },
        )

    if archive_name_source is not None and archive_name_source not in ("metadata", "filename"):
        return JSONResponse(
            status_code=400,
            content={"detail": "archive_name_source must be 'metadata' or 'filename'"},
        )

    if tailscale_disabled is not None:
        await set_setting(db, "virtual_printer_tailscale_disabled", "true" if tailscale_disabled else "false")
        # The first virtual printer's card shows its Tailscale address by it.
        if first_vp is not None:
            first_vp.tailscale_disabled = tailscale_disabled
    if archive_name_source is not None:
        await set_setting(db, "virtual_printer_archive_name_source", archive_name_source)

    await db.commit()
    db.expire_all()

    return await get_virtual_printer_settings(db, printer_scope=printer_scope)


# =============================================================================
# MQTT Relay Settings
# =============================================================================


@router.get("/mqtt/status")
async def get_mqtt_status(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Get MQTT relay connection status."""
    from backend.app.services.mqtt_relay import mqtt_relay

    return mqtt_relay.get_status()