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

import logging
import os
from pathlib import Path

from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.auth import (
    ApiKeyActor,
    RequestActor,
    RequestPrinterScope,
    RequirePermissionIfAuthEnabled,
    get_current_active_user,
)
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.user import User
from backend.app.schemas.not_null import IgnoresNullForRequiredColumns
from backend.app.schemas.virtual_printer import VPDiagnosticResult

# Imported at module scope so tests can patch
# backend.app.api.routes.virtual_printers.tailscale_service.
from backend.app.services.virtual_printer.tailscale import tailscale_service

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/virtual-printers", tags=["virtual-printers"])


class TailscaleStatusResponse(BaseModel):
    available: bool
    fqdn: str
    hostname: str
    tailnet_name: str
    tailscale_ips: list[str]
    error: str | None


class VirtualPrinterCreate(BaseModel):
    name: str = "Bambuddy"
    enabled: bool = False
    mode: str = "library"
    model: str | None = None
    access_code: str | None = None
    target_printer_id: int | None = None
    library_folder_id: int | None = None
    queue_remove_after_print: bool = False
    auto_dispatch: bool = True
    queue_force_color_match: bool = False
    save_ams_mapping: bool = False
    gcode_injection: bool = False
    bind_ip: str | None = None
    remote_interface_ip: str | None = None


class VirtualPrinterUpdate(IgnoresNullForRequiredColumns):
    __orm_table__ = "virtual_printers"
    name: str | None = None
    enabled: bool | None = None
    mode: str | None = None
    model: str | None = None
    access_code: str | None = None
    target_printer_id: int | None = None
    library_folder_id: int | None = None
    queue_remove_after_print: bool | None = None
    auto_dispatch: bool | None = None
    queue_force_color_match: bool | None = None
    save_ams_mapping: bool | None = None
    gcode_injection: bool | None = None
    bind_ip: str | None = None
    remote_interface_ip: str | None = None
    tailscale_disabled: bool | None = None


def _resolve_printer_model(printer_model: str | None, serial: str | None = None) -> str | None:
    """Map a printer's model (display name or SSDP code) to a valid VP SSDP model code.

    Printers store display names like 'X1C' while VPs need SSDP codes like
    'BL-P001'. ``serial`` tells apart the codes one name covers (H2C: O1C or
    O1C2); see ``manager.model_code_for_printer``.
    """
    from backend.app.services.virtual_printer.manager import model_code_for_printer

    return model_code_for_printer(printer_model, serial)


async def _check_upload_folder(db: AsyncSession, folder_id: int, user) -> JSONResponse | None:
    """None when the virtual printer may store uploads in ``folder_id``.

    The user setting it must be able to add files there themselves (#3201),
    and it must be an internal folder: an external mount refuses a second file
    of the same name, and slicer names repeat.
    """
    from backend.app.services.library_folder_access import get_writable_folder

    folder = await get_writable_folder(db, folder_id, user)
    if folder.is_external:
        return JSONResponse(
            status_code=400,
            content={"detail": "Virtual printer uploads can't go to an external folder"},
        )
    return None


def _target_missing(target_printer_id: int) -> JSONResponse:
    return JSONResponse(status_code=400, content={"detail": f"Printer with ID {target_printer_id} not found"})


def _vp_missing() -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": "Virtual printer not found"})


async def _vp_to_dict(vp, db: AsyncSession, status: dict | None = None) -> dict:
    """Convert VirtualPrinter model to response dict.

    In proxy mode the surfaced serial is the target printer's actual serial
    (what the bridge advertises over SSDP / what slicers see), not the
    self-generated suffix. Archive / queue / review keep the self-generated
    serial since those modes never speak the target's identity.
    """
    from backend.app.models.printer import Printer
    from backend.app.models.virtual_printer import VP_MODE_PROXY
    from backend.app.services.virtual_printer import VIRTUAL_PRINTER_MODELS
    from backend.app.services.virtual_printer.manager import DEFAULT_VIRTUAL_PRINTER_MODEL, _get_serial_for_model

    model_code = vp.model or DEFAULT_VIRTUAL_PRINTER_MODEL
    serial = _get_serial_for_model(model_code, vp.serial_suffix)
    if vp.mode == VP_MODE_PROXY and vp.target_printer_id:
        result = await db.execute(select(Printer.serial_number).where(Printer.id == vp.target_printer_id))
        target_serial = result.scalar_one_or_none()
        if target_serial:
            serial = target_serial

    return {
        "id": vp.id,
        "name": vp.name,
        "enabled": vp.enabled,
        "mode": vp.mode,
        "model": model_code,
        "model_name": VIRTUAL_PRINTER_MODELS.get(model_code, model_code),
        "access_code_set": bool(vp.access_code),
        "serial": serial,
        "target_printer_id": vp.target_printer_id,
        "library_folder_id": vp.library_folder_id,
        "queue_remove_after_print": vp.queue_remove_after_print,
        "auto_dispatch": vp.auto_dispatch,
        "queue_force_color_match": vp.queue_force_color_match,
        "save_ams_mapping": vp.save_ams_mapping,
        "gcode_injection": vp.gcode_injection,
        "bind_ip": vp.bind_ip,
        "remote_interface_ip": vp.remote_interface_ip,
        "tailscale_disabled": vp.tailscale_disabled,
        "position": vp.position,
        "status": status or {"running": False, "pending_files": 0},
    }


@router.get("")
async def list_virtual_printers(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """List all virtual printers with status."""
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import VIRTUAL_PRINTER_MODELS, virtual_printer_manager

    result = await db.execute(select(VirtualPrinter).order_by(VirtualPrinter.position, VirtualPrinter.id))
    vps = result.scalars().all()

    printers = []
    for vp in vps:
        # One bridging to a printer the caller can't use is left out: its
        # status names that printer.
        if not printer_scope.allows(vp.target_printer_id):
            continue
        instance = virtual_printer_manager.get_instance(vp.id)
        status = instance.get_status() if instance else {"running": False, "pending_files": 0}
        printers.append(await _vp_to_dict(vp, db, status))

    from backend.app.services.virtual_printer.manager import MODEL_SERIAL_PREFIXES

    return {
        "printers": printers,
        "models": VIRTUAL_PRINTER_MODELS,
        # The serial family each model's VP gets (Bambu's first 3 characters),
        # which tells apart models of one name (H2C: O1C 094, O1C2 31B).
        "model_serial_prefixes": {code: prefix[:3] for code, prefix in MODEL_SERIAL_PREFIXES.items()},
    }


@router.post("")
async def create_virtual_printer(
    body: VirtualPrinterCreate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
    # Who the #3201 folder check runs against: an API key as its owner.
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Create a new virtual printer."""
    from backend.app.models.virtual_printer import VP_MODE_VALUES, VirtualPrinter, normalize_vp_mode
    from backend.app.services.virtual_printer import VIRTUAL_PRINTER_MODELS, virtual_printer_manager
    from backend.app.services.virtual_printer.manager import DEFAULT_VIRTUAL_PRINTER_MODEL

    # Accept both canonical and legacy wire values so older clients (forks /
    # mobile shortcuts / scripted setups) still work; normalize before write.
    body.mode = normalize_vp_mode(body.mode) or body.mode
    if body.mode not in VP_MODE_VALUES:
        return JSONResponse(status_code=400, content={"detail": "Invalid mode"})

    # Validate model
    if body.model and body.model not in VIRTUAL_PRINTER_MODELS:
        return JSONResponse(
            status_code=400,
            content={"detail": f"Invalid model. Must be one of: {', '.join(VIRTUAL_PRINTER_MODELS.keys())}"},
        )

    # Validate access code length
    if body.access_code and len(body.access_code) != 8:
        return JSONResponse(status_code=400, content={"detail": "Access code must be exactly 8 characters"})

    # Validation when enabling. Non-proxy VPs with a target printer derive
    # their access code from the target (the bridge forwards the slicer's
    # auth bytes through to the real printer, so the codes MUST match),
    # so a separately-supplied access_code isn't required in that case.
    if body.enabled:
        if not body.bind_ip:
            return JSONResponse(status_code=400, content={"detail": "Bind IP is required when enabling"})
        if body.mode == "proxy":
            if not body.target_printer_id:
                return JSONResponse(status_code=400, content={"detail": "Target printer is required for proxy mode"})
        else:
            if not body.access_code and not body.target_printer_id:
                return JSONResponse(status_code=400, content={"detail": "Access code is required when enabling"})

    # Validate proxy target printer exists
    target_printer = None
    if body.target_printer_id is not None:
        from backend.app.models.printer import Printer

        result = await db.execute(select(Printer).where(Printer.id == body.target_printer_id))
        target_printer = result.scalar_one_or_none()
        if not target_printer or not printer_scope.allows(target_printer.id):
            return _target_missing(body.target_printer_id)

    if body.library_folder_id is not None and (
        refused := await _check_upload_folder(db, body.library_folder_id, actor)
    ):
        return refused

    # Validate bind_ip uniqueness (against all enabled VPs)
    if body.bind_ip:
        result = await db.execute(
            select(VirtualPrinter).where(
                VirtualPrinter.bind_ip == body.bind_ip,
                VirtualPrinter.enabled == True,  # noqa: E712
            )
        )
        if result.scalar_one_or_none():
            return JSONResponse(status_code=400, content={"detail": f"Bind IP {body.bind_ip} is already in use"})

    # Force-inherit the access code from the target printer for non-proxy VPs.
    # The non-proxy bridge (Immediate / Review / Queue with a target set) forwards
    # the slicer's MQTT / RTSPS auth bytes through to the real printer, so any
    # value the user supplied here would silently break the bridge if it didn't
    # match the printer's code. The UI now renders the field read-only when a
    # target is set; this is the belt-and-braces backstop for any non-UI client.
    effective_access_code = body.access_code
    if body.mode != "proxy" and target_printer is not None:
        effective_access_code = target_printer.access_code

    # Generate next serial suffix
    result = await db.execute(select(VirtualPrinter.serial_suffix).order_by(VirtualPrinter.id.desc()))
    last_suffix = result.scalar()
    if last_suffix:
        try:
            next_num = int(last_suffix) + 1
            new_suffix = str(next_num).zfill(9)
        except ValueError:
            new_suffix = "391800002"
    else:
        new_suffix = "391800001"

    # Get next position
    result = await db.execute(select(VirtualPrinter.position).order_by(VirtualPrinter.position.desc()))
    last_pos = result.scalar()
    next_pos = (last_pos or 0) + 1

    vp = VirtualPrinter(
        name=body.name,
        enabled=body.enabled,
        mode=body.mode,
        model=body.model
        or _resolve_printer_model(
            target_printer.model if target_printer and body.mode == "proxy" else None,
            target_printer.serial_number if target_printer else None,
        )
        or DEFAULT_VIRTUAL_PRINTER_MODEL,
        access_code=effective_access_code,
        target_printer_id=body.target_printer_id,
        library_folder_id=body.library_folder_id,
        queue_remove_after_print=body.queue_remove_after_print,
        auto_dispatch=body.auto_dispatch,
        queue_force_color_match=body.queue_force_color_match,
        save_ams_mapping=body.save_ams_mapping,
        gcode_injection=body.gcode_injection,
        bind_ip=body.bind_ip,
        remote_interface_ip=body.remote_interface_ip,
        serial_suffix=new_suffix,
        position=next_pos,
    )
    db.add(vp)
    await db.commit()
    await db.refresh(vp)

    logger.info("Created virtual printer: %s (id=%d)", vp.name, vp.id)

    # Sync services if enabled
    if body.enabled:
        try:
            await virtual_printer_manager.sync_from_db()
        except Exception as e:
            logger.error("Failed to start virtual printer after create: %s", e)

    return await _vp_to_dict(vp, db)


@router.get("/tailscale-status", response_model=TailscaleStatusResponse)
async def get_tailscale_status(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
) -> TailscaleStatusResponse:
    """Return current Tailscale availability and machine identity.

    Used by the frontend to indicate whether virtual printer TLS is backed
    by a trusted Let's Encrypt certificate or a self-signed CA.
    """
    status = await tailscale_service.get_status()
    return TailscaleStatusResponse(
        available=status.available,
        fqdn=status.fqdn,
        hostname=status.hostname,
        tailnet_name=status.tailnet_name,
        tailscale_ips=status.tailscale_ips,
        error=status.error,
    )


@router.get("/ca-certificate")
async def get_ca_certificate(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Return the shared virtual-printer CA certificate (PEM) for slicer trust import.

    One CA is shared by every virtual printer — the user imports it into their
    slicer's trust store once. Only the public certificate is returned; the CA
    private key never leaves the backend.
    """
    from backend.app.services.virtual_printer import virtual_printer_manager

    try:
        return virtual_printer_manager.get_ca_certificate_info()
    except Exception as e:
        logger.error("Failed to obtain virtual printer CA certificate: %s", e)
        return JSONResponse(status_code=500, content={"detail": "Could not generate the CA certificate"})


@router.get("/ca-certificate/installer/{platform}")
async def get_ca_certificate_installer(
    platform: str,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
    """Return a script that adds the CA to Bambu Studio and OrcaSlicer.

    ``platform`` is ``windows`` (a .cmd) or ``unix`` (a .sh for Linux and
    macOS). The CA is written into the script, which then needs no access to
    Bambuddy. Returns ``{"filename", "content"}``.
    """
    from backend.app.services.virtual_printer import virtual_printer_manager
    from backend.app.services.virtual_printer.slicer_cert_installer import (
        INSTALLER_PLATFORMS,
        build_slicer_cert_installer,
    )

    if platform not in INSTALLER_PLATFORMS:
        return JSONResponse(status_code=404, content={"detail": f"Unknown platform '{platform}'"})
    try:
        info = virtual_printer_manager.get_ca_certificate_info()
    except Exception as e:
        logger.error("Failed to obtain virtual printer CA certificate: %s", e)
        return JSONResponse(status_code=500, content={"detail": "Could not generate the CA certificate"})
    installer = build_slicer_cert_installer(platform, info["pem"], info["fingerprint_sha256"])
    return {"filename": installer.filename, "content": installer.content}


@router.get("/my-access-code")
async def get_my_access_code(
    current_user: User = Depends(get_current_active_user),
    db: AsyncSession = Depends(get_db, scope="function"),
):
    """The signed-in user's personal access code for virtual printers (#1430).

    Entered in their slicer instead of a virtual printer's own code, it marks
    what they send as theirs. Made on first use. Only for a signed-in user (an
    API key has no slicer); a virtual printer checks at every sign-in whether
    the user may send to it.
    """
    from backend.app.services.virtual_printer.user_codes import code_for

    user = await db.get(User, current_user.id)
    if user is None:
        raise HTTPException(status_code=401, detail="Not signed in")
    return {"code": await code_for(db, user)}


@router.post("/my-access-code")
async def renew_my_access_code(
    current_user: User = Depends(get_current_active_user),
    db: AsyncSession = Depends(get_db, scope="function"),
):
    """Replace the signed-in user's personal access code; the old one stops working."""
    from backend.app.services.virtual_printer.user_codes import code_for

    user = await db.get(User, current_user.id)
    if user is None:
        raise HTTPException(status_code=401, detail="Not signed in")
    return {"code": await code_for(db, user, renew=True)}


@router.get("/{vp_id}/diagnostic", response_model=VPDiagnosticResult)
async def diagnose_virtual_printer(
    vp_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Run setup diagnostics for a virtual printer.

    Probes the VP's own bind IP and services so the user can self-diagnose the
    common "my virtual printer doesn't show up in the slicer" failures.
    """
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import virtual_printer_manager
    from backend.app.services.virtual_printer.diagnostic import run_vp_diagnostic

    result = await db.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))
    vp = result.scalar_one_or_none()
    if not vp or not printer_scope.allows(vp.target_printer_id):
        return _vp_missing()

    instance = virtual_printer_manager.get_instance(vp.id)
    return await run_vp_diagnostic(vp, instance)


@router.get("/{vp_id}")
async def get_virtual_printer(
    vp_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get a single virtual printer with status."""
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import virtual_printer_manager

    result = await db.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))
    vp = result.scalar_one_or_none()
    if not vp or not printer_scope.allows(vp.target_printer_id):
        return _vp_missing()

    instance = virtual_printer_manager.get_instance(vp.id)
    status = instance.get_status() if instance else {"running": False, "pending_files": 0}

    return await _vp_to_dict(vp, db, status)


@router.put("/{vp_id}")
async def update_virtual_printer(
    vp_id: int,
    body: VirtualPrinterUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
    # Who the #3201 folder check runs against: an API key as its owner.
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Update a virtual printer."""
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import VIRTUAL_PRINTER_MODELS, virtual_printer_manager

    result = await db.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))
    vp = result.scalar_one_or_none()
    # One bridging to a printer the caller can't use is not theirs to change.
    if not vp or not printer_scope.allows(vp.target_printer_id):
        return _vp_missing()

    # Redact the access code before logging — model_dump otherwise includes
    # the plaintext value at DEBUG, violating the project no-secrets-in-logs
    # rule. Replace with a marker that still signals "the user changed it"
    # vs "the user didn't touch this field".
    _safe_body = body.model_dump(exclude_unset=True)
    if "access_code" in _safe_body:
        _safe_body["access_code"] = "***"
    logger.debug(
        "Update VP %d: body=%s, current state: mode=%s, enabled=%s, access_code_set=%s, bind_ip=%s, target=%s",
        vp_id,
        _safe_body,
        vp.mode,
        vp.enabled,
        bool(vp.access_code),
        vp.bind_ip,
        vp.target_printer_id,
    )

    # Apply updates
    if body.name is not None:
        vp.name = body.name
    if body.mode is not None:
        from backend.app.models.virtual_printer import VP_MODE_VALUES, normalize_vp_mode

        canonical_mode = normalize_vp_mode(body.mode) or body.mode
        if canonical_mode not in VP_MODE_VALUES:
            return JSONResponse(status_code=400, content={"detail": "Invalid mode"})
        vp.mode = canonical_mode
    if body.model is not None:
        if body.model not in VIRTUAL_PRINTER_MODELS:
            return JSONResponse(
                status_code=400,
                content={"detail": f"Invalid model. Must be one of: {', '.join(VIRTUAL_PRINTER_MODELS.keys())}"},
            )
        vp.model = body.model
    if body.access_code is not None:
        if body.access_code and len(body.access_code) != 8:
            return JSONResponse(status_code=400, content={"detail": "Access code must be exactly 8 characters"})
        vp.access_code = body.access_code
    if "target_printer_id" in body.model_fields_set and body.target_printer_id is None:
        # An explicit null clears the target: a queue VP then sends its jobs
        # to any printer of its model. The access code stays what it was, so
        # a slicer already bound keeps working. A proxy VP needs a target, so
        # the enable checks below turn it off.
        vp.target_printer_id = None
    elif body.target_printer_id is not None:
        from backend.app.models.printer import Printer

        result = await db.execute(select(Printer).where(Printer.id == body.target_printer_id))
        target_printer = result.scalar_one_or_none()
        if not target_printer or not printer_scope.allows(target_printer.id):
            return _target_missing(body.target_printer_id)
        vp.target_printer_id = body.target_printer_id
        # Auto-inherit model from target printer in proxy mode (unless user explicitly set model)
        if body.model is None and vp.mode == "proxy" and target_printer.model:
            vp.model = (
                _resolve_printer_model(target_printer.model, target_printer.serial_number) or target_printer.model
            )
    if "library_folder_id" in body.model_fields_set and body.library_folder_id is None:
        vp.library_folder_id = None  # back to the shared "Virtual Printer" folder
    elif body.library_folder_id is not None:
        if refused := await _check_upload_folder(db, body.library_folder_id, actor):
            return refused
        vp.library_folder_id = body.library_folder_id
    if body.queue_remove_after_print is not None:
        vp.queue_remove_after_print = body.queue_remove_after_print
    if body.auto_dispatch is not None:
        vp.auto_dispatch = body.auto_dispatch
    if body.queue_force_color_match is not None:
        vp.queue_force_color_match = body.queue_force_color_match
    if body.save_ams_mapping is not None:
        vp.save_ams_mapping = body.save_ams_mapping
    if body.gcode_injection is not None:
        vp.gcode_injection = body.gcode_injection
    if body.bind_ip is not None:
        vp.bind_ip = body.bind_ip
    if body.remote_interface_ip is not None:
        vp.remote_interface_ip = body.remote_interface_ip
    if body.tailscale_disabled is not None:
        vp.tailscale_disabled = body.tailscale_disabled

    # Auto-inherit model when switching to proxy mode with existing target printer
    if body.mode == "proxy" and body.model is None and body.target_printer_id is None and vp.target_printer_id:
        from backend.app.models.printer import Printer as PrinterModel

        result = await db.execute(select(PrinterModel).where(PrinterModel.id == vp.target_printer_id))
        existing_target = result.scalar_one_or_none()
        if existing_target and existing_target.model:
            vp.model = (
                _resolve_printer_model(existing_target.model, existing_target.serial_number) or existing_target.model
            )

    # Force-inherit the access code from the target printer for non-proxy VPs.
    # See create_virtual_printer for the rationale: the bridge forwards slicer
    # auth bytes through, so the VP's code MUST equal the target's. This block
    # runs after every patch (whether or not access_code or target were in the
    # body), so changing the target also resyncs the code, and an explicit
    # access_code submitted alongside a target is silently overridden.
    if vp.mode != "proxy" and vp.target_printer_id is not None:
        from backend.app.models.printer import Printer as PrinterModelAC

        result = await db.execute(select(PrinterModelAC).where(PrinterModelAC.id == vp.target_printer_id))
        target_for_ac = result.scalar_one_or_none()
        if target_for_ac is not None and vp.access_code != target_for_ac.access_code:
            vp.access_code = target_for_ac.access_code

    # Determine final enabled state
    explicitly_enabling = body.enabled is True
    new_enabled = body.enabled if body.enabled is not None else vp.enabled
    effective_mode = vp.mode

    if explicitly_enabling:
        # User is explicitly toggling on — enforce all requirements
        if not vp.bind_ip:
            logger.warning("Update VP %d rejected: no bind_ip", vp_id)
            return JSONResponse(status_code=400, content={"detail": "Bind IP is required when enabling"})
        # Validate bind_ip uniqueness (against all enabled VPs)
        existing = await db.execute(
            select(VirtualPrinter).where(
                VirtualPrinter.bind_ip == vp.bind_ip,
                VirtualPrinter.id != vp_id,
                VirtualPrinter.enabled == True,  # noqa: E712
            )
        )
        conflict = existing.scalar_one_or_none()
        if conflict:
            logger.warning(
                "Update VP %d rejected: bind_ip %s already in use by VP %d (enabled=%s, mode=%s)",
                vp_id,
                vp.bind_ip,
                conflict.id,
                conflict.enabled,
                conflict.mode,
            )
            return JSONResponse(
                status_code=400,
                content={
                    "detail": f"Bind IP {vp.bind_ip} is already in use by '{conflict.name}'"
                    if printer_scope.allows(conflict.target_printer_id)
                    else f"Bind IP {vp.bind_ip} is already in use"
                },
            )
        if effective_mode == "proxy":
            if not vp.target_printer_id:
                logger.warning("Update VP %d rejected: no target_printer_id for proxy mode", vp_id)
                return JSONResponse(status_code=400, content={"detail": "Target printer is required for proxy mode"})
        else:
            if not vp.access_code:
                logger.warning(
                    "Update VP %d rejected: no access_code for non-proxy enable (mode=%s)", vp_id, effective_mode
                )
                return JSONResponse(status_code=400, content={"detail": "Access code is required when enabling"})
    elif new_enabled and body.enabled is None:
        # VP is already enabled and user is changing other fields —
        # auto-disable if new state doesn't meet requirements
        if not vp.bind_ip:
            new_enabled = False
        elif effective_mode == "proxy":
            if not vp.target_printer_id:
                new_enabled = False
        else:
            if not vp.access_code:
                new_enabled = False

    vp.enabled = new_enabled

    await db.commit()
    await db.refresh(vp)

    logger.info("Updated virtual printer: %s (id=%d)", vp.name, vp.id)

    # Sync services
    try:
        await virtual_printer_manager.sync_from_db()
    except Exception as e:
        logger.error("Failed to sync virtual printers after update: %s", e)

    instance = virtual_printer_manager.get_instance(vp.id)
    status = instance.get_status() if instance else {"running": False, "pending_files": 0}

    return await _vp_to_dict(vp, db, status)


async def drop_virtual_printer(db: AsyncSession, vp_id: int) -> Path:
    """Delete a virtual printer's row and discard its pending uploads.

    Its instance must be stopped first, if one runs. The caller commits, then
    calls remove_upload_dir with what this returns. *db* is a session or a
    connection.
    """
    from sqlalchemy import delete as sql_delete

    from backend.app.models.pending_upload import PendingUpload
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import virtual_printer_manager

    upload_dir = virtual_printer_manager._base_dir / "uploads" / str(vp_id)
    # Mark any PendingUpload rows that referenced this VP's upload_dir as
    # discarded — without this the rows live on as phantom entries in
    # /pending-uploads/ pointing at file paths that no longer exist, and
    # the user only learns they're orphaned by trying to archive one and
    # getting a flip-to-discarded on file-missing.
    try:
        await db.execute(
            update(PendingUpload)
            # The folder, not every path that starts with its name (uploads/12 for 1).
            .where(PendingUpload.file_path.startswith(f"{upload_dir}{os.sep}", autoescape=True))
            .values(status="discarded")
        )
    except Exception as e:
        logger.error("Failed to discard orphan PendingUpload rows for VP %d: %s", vp_id, e)
    await db.execute(sql_delete(VirtualPrinter).where(VirtualPrinter.id == vp_id))
    return upload_dir


def remove_upload_dir(upload_dir: Path) -> None:
    """Remove a deleted virtual printer's upload directory.

    After the DB commit succeeds, so a crash between commit and rmtree only
    leaves orphan files (vs orphan rows pointing at a now-missing tree).
    """
    if upload_dir.exists():
        import shutil

        shutil.rmtree(upload_dir, ignore_errors=True)


@router.delete("/{vp_id}")
async def delete_virtual_printer(
    vp_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Delete a virtual printer."""
    from backend.app.models.virtual_printer import VirtualPrinter
    from backend.app.services.virtual_printer import virtual_printer_manager

    result = await db.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))
    vp = result.scalar_one_or_none()
    if not vp or not printer_scope.allows(vp.target_printer_id):
        return _vp_missing()

    vp_name = vp.name

    # Stop instance if running
    await virtual_printer_manager.remove_instance(vp_id)
    upload_dir = await drop_virtual_printer(db, vp_id)
    await db.commit()
    remove_upload_dir(upload_dir)

    logger.info("Deleted virtual printer: %s (id=%d)", vp_name, vp_id)

    # Resync remaining services
    try:
        await virtual_printer_manager.sync_from_db()
    except Exception as e:
        logger.error("Failed to sync virtual printers after delete: %s", e)

    return {"detail": "Deleted", "id": vp_id}