bambuddy/backend/app/services/printer_media.py

"""Helpers for matching and downloading printer-side video files."""

from __future__ import annotations

import asyncio
import json
import logging
import re
import secrets
import shutil
import tempfile
import time
import zipfile
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from pathlib import Path, PurePosixPath

from backend.app.core.config import settings
from backend.app.core.tasks import spawn_background_task
from backend.app.services.bambu_ftp import (
    DownloadCancelled,
    DownloadInsufficientSpace,
    DownloadLimitExceeded,
    download_file_async,
)

logger = logging.getLogger(__name__)

VIDEO_SUFFIXES = (".mp4", ".avi", ".mkv")
MAX_PRINTER_ZIP_BYTES = 10 * 1024**3
PRINTER_ZIP_FREE_SPACE_RESERVE = 256 * 1024**2
_STALE_BUNDLE_SECONDS = 60 * 60
MAX_PRINTER_ZIP_PREPARE_SECONDS = 30 * 60
MAX_OPEN_ARCHIVE_IPCAM_SECONDS = 24 * 60 * 60
_BUNDLE_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{1,200}$")
_JOB_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{20,200}$")
_LOCAL_JOB_TASKS: dict[str, asyncio.Task] = {}
_cleanup_task: asyncio.Task | None = None
_CLEANUP_INTERVAL_SECONDS = 15 * 60


class PrinterFilesZipTooLargeError(ValueError):
    """The selected printer files exceed the bounded ZIP staging limit."""


class PrinterFilesZipInsufficientSpaceError(OSError):
    """The app data volume cannot safely stage the selected files."""


@dataclass(frozen=True)
class PrinterFilesZipResult:
    """Result of staging one printer ZIP."""

    path: Path
    requested: int
    successful: int
    failed_paths: tuple[str, ...]
    total_bytes: int


@dataclass(frozen=True)
class PrinterFilesJobStatus:
    """Serializable state for an asynchronous browser preparation job."""

    job_id: str
    printer_id: int
    state: str
    requested: int
    successful: int = 0
    failed: int = 0
    token: str | None = None
    filename: str | None = None
    message: str | None = None


class _FileCancelSignal:
    """Cross-worker cancellation signal checked by the FTP callback thread."""

    def __init__(self, path: Path):
        self.path = path
        self._last_check = 0.0
        self._cached = False

    def is_set(self) -> bool:
        if self._cached:
            return True
        now = time.monotonic()
        if now - self._last_check >= 0.25:
            self._last_check = now
            self._cached = self.path.exists()
        return self._cached


def _job_status_path(job_id: str) -> Path:
    if not _JOB_KEY_RE.fullmatch(job_id):
        raise ValueError("Invalid printer download job id")
    return _printer_zip_root() / f"job-{job_id}.json"


def _job_cancel_path(job_id: str) -> Path:
    if not _JOB_KEY_RE.fullmatch(job_id):
        raise ValueError("Invalid printer download job id")
    return _printer_zip_root() / f"job-{job_id}.cancel"


def _write_job_status(status: PrinterFilesJobStatus) -> None:
    """Atomically publish job state for polling from any app worker."""

    path = _job_status_path(status.job_id)
    temp_path = path.with_suffix(".tmp")
    temp_path.write_text(json.dumps(status.__dict__, separators=(",", ":")), encoding="utf-8")
    temp_path.replace(path)


def _read_job_status(job_id: str) -> PrinterFilesJobStatus | None:
    try:
        data = json.loads(_job_status_path(job_id).read_text(encoding="utf-8"))
        return PrinterFilesJobStatus(**data)
    except (FileNotFoundError, OSError, ValueError, TypeError, json.JSONDecodeError):
        return None


def _naive_utc(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is not None:
        return value.astimezone(timezone.utc).replace(tzinfo=None)
    return value


def match_ipcam_chunks(
    files: list[dict],
    started_at: datetime | None,
    completed_at: datetime | None,
    *,
    now: datetime | None = None,
) -> list[dict]:
    """Return `/ipcam` chunks whose completion time overlaps a print.

    Bambu's `ipcam-record.*.mp4` files are fixed-size chunks. On the tested X1C
    and H2D firmware, their FTP mtime is the chunk completion time in the same
    UTC-naive basis used by archive timestamps. Some firmware reports FTP LIST
    mtimes in printer-local time instead; LIST carries no timezone with which
    to correct those values reliably. A ten-minute tail includes the final
    chunk, whose mtime lands after the print-complete event.
    """

    start = _naive_utc(started_at)
    if start is None:
        return []
    live_end = _naive_utc(now) or datetime.now(timezone.utc).replace(tzinfo=None)
    # A crash can leave an archive in ``printing`` indefinitely. Do not turn
    # that stale row into a window covering every chunk created since then.
    end = _naive_utc(completed_at) or min(live_end, start + timedelta(seconds=MAX_OPEN_ARCHIVE_IPCAM_SECONDS))
    lower = start - timedelta(minutes=1)
    upper = max(start, end) + timedelta(minutes=10)

    matches: list[dict] = []
    for file in files:
        name = str(file.get("name") or "")
        mtime = file.get("mtime")
        if file.get("is_directory") or not name.lower().startswith("ipcam-record."):
            continue
        if not name.lower().endswith(VIDEO_SUFFIXES) or not isinstance(mtime, datetime):
            continue
        timestamp = _naive_utc(mtime)
        if timestamp is not None and lower <= timestamp <= upper:
            matches.append(file)

    matches.sort(key=lambda item: _naive_utc(item.get("mtime")) or datetime.min)
    return matches


def _zip_arcname(remote_path: str, used: set[str]) -> str:
    """Return a safe, unique relative archive name for a printer path."""

    parts = [part for part in PurePosixPath(remote_path).parts if part not in ("/", "", ".", "..")]
    candidate = "/".join(parts) or "printer-file"
    stem = candidate
    suffix = ""
    if "." in PurePosixPath(candidate).name:
        suffix = "".join(PurePosixPath(candidate).suffixes)
        stem = candidate[: -len(suffix)] if suffix else candidate
    counter = 2
    while candidate in used:
        candidate = f"{stem}-{counter}{suffix}"
        counter += 1
    used.add(candidate)
    return candidate


def _printer_zip_root() -> Path:
    """Return the dedicated staging root without doing event-loop I/O."""

    return settings.archive_dir / "temp" / "printer-file-downloads"


def _ensure_printer_zip_root() -> Path:
    """Create and return the staging root on the persistent data volume."""

    root = _printer_zip_root()
    root.mkdir(parents=True, exist_ok=True)
    return root


def _prune_stale_bundles(root: Path) -> None:
    """Remove abandoned bundles after token expiry, without touching archives."""

    cutoff = time.time() - _STALE_BUNDLE_SECONDS
    if not root.exists():
        return
    for child in root.iterdir():
        try:
            if child.is_dir() and child.stat().st_mtime < cutoff:
                shutil.rmtree(child, ignore_errors=True)
            elif child.is_file() and child.name.startswith("job-") and child.stat().st_mtime < cutoff:
                child.unlink(missing_ok=True)
        except OSError:
            continue


async def prune_stale_printer_file_bundles() -> None:
    """Prune abandoned printer ZIPs without blocking the event loop."""

    root = await asyncio.to_thread(_ensure_printer_zip_root)
    await asyncio.to_thread(_prune_stale_bundles, root)


async def _printer_download_cleanup_loop() -> None:
    while True:
        try:
            await asyncio.sleep(_CLEANUP_INTERVAL_SECONDS)
            await prune_stale_printer_file_bundles()
        except asyncio.CancelledError:
            break
        except Exception:
            logger.exception("Periodic printer-download cleanup failed")


def start_printer_download_cleanup() -> None:
    global _cleanup_task
    if _cleanup_task is None:
        _cleanup_task = spawn_background_task(_printer_download_cleanup_loop(), name="printer-download-cleanup")


async def stop_printer_download_cleanup() -> None:
    """Stop cleanup and cancel every in-process preparation before shutdown."""

    global _cleanup_task
    tasks: list[asyncio.Task] = []
    cleanup_task = _cleanup_task
    _cleanup_task = None
    if cleanup_task is not None:
        cleanup_task.cancel()
        tasks.append(cleanup_task)

    # Jobs can be inside an FTP worker thread. Publish the same cooperative
    # cancellation marker used by the DELETE endpoint before cancelling the
    # asyncio wrapper, then await every wrapper so no executor work is left
    # behind when the application event loop closes.
    for job_id, task in list(_LOCAL_JOB_TASKS.items()):
        if not task.done():
            await asyncio.to_thread(_job_cancel_path(job_id).touch)
            task.cancel()
        tasks.append(task)
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)
    _LOCAL_JOB_TASKS.clear()


def printer_files_zip_path(printer_id: int, token: str) -> Path | None:
    """Resolve the staged ZIP for a resource-bound browser token."""

    bundle_key = f"{printer_id}-{token}"
    if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
        return None
    return _printer_zip_root() / bundle_key / "printer-files.zip"


def bind_printer_files_zip_to_token(
    result: PrinterFilesZipResult,
    printer_id: int,
    token: str,
) -> PrinterFilesZipResult:
    """Move a prepared bundle to the path derived from its persisted token."""

    target = printer_files_zip_path(printer_id, token)
    if target is None:
        raise ValueError("Invalid printer ZIP token")
    result.path.parent.rename(target.parent)
    return replace(result, path=target)


def _check_initial_space(root: Path, sizes: dict[str, int]) -> None:
    # These sizes are client-reported hints used only for an early rejection,
    # so this is a courtesy, not the bound. The real one is enforced per write
    # and per FTP callback below, against actual bytes and the live free space,
    # which is the only thing that can hold when several preparations run at
    # once -- and they do: nothing serializes them. Two concurrent jobs that
    # both pass here stop independently at the reserve, and the one that gets
    # there second fails with a message saying so.
    expected_total = sum(sizes.values())
    if expected_total > MAX_PRINTER_ZIP_BYTES:
        raise PrinterFilesZipTooLargeError(
            f"Selected files total {expected_total} bytes; the limit is {MAX_PRINTER_ZIP_BYTES} bytes"
        )

    largest_file = max(sizes.values(), default=0)
    # In the worst case the ZIP is as large as the inputs while the largest
    # source is still staged beside it. Keep a reserve for the database/logs.
    required = expected_total + largest_file + PRINTER_ZIP_FREE_SPACE_RESERVE
    free = shutil.disk_usage(root).free
    if free < required:
        raise PrinterFilesZipInsufficientSpaceError(
            f"The app data volume needs {required} bytes free to stage this selection; {free} bytes are available"
        )


async def build_printer_files_zip(
    printer,
    paths: list[str],
    sizes: dict[str, int],
    *,
    bundle_key: str | None = None,
    preserve_paths: bool = True,
    allow_empty: bool = False,
    cancel_signal: _FileCancelSignal | None = None,
    progress_callback: Callable[[int, int], Awaitable[None]] | None = None,
) -> PrinterFilesZipResult:
    """Download printer files one at a time into a disk-backed ZIP.

    The previous implementation held every source file and the final ZIP in
    memory. Continuous `/ipcam` chunks are commonly ~250 MB each, so selecting
    only a few could exhaust both server and browser memory.
    """

    root = await asyncio.to_thread(_ensure_printer_zip_root)
    await asyncio.to_thread(_prune_stale_bundles, root)
    await asyncio.to_thread(_check_initial_space, root, sizes)
    bundle_dir: Path | None = None
    try:
        if bundle_key is None:
            bundle_dir = Path(await asyncio.to_thread(tempfile.mkdtemp, prefix="bundle-", dir=root))
        else:
            if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
                raise ValueError("Invalid printer ZIP bundle key")
            bundle_dir = root / bundle_key
            await asyncio.to_thread(bundle_dir.mkdir, mode=0o700)
        zip_path = bundle_dir / "printer-files.zip"
        successful = 0
        total_bytes = 0
        failed_paths: list[str] = []
        used_names: set[str] = set()

        archive = await asyncio.to_thread(zipfile.ZipFile, zip_path, "w", allowZip64=True)
        try:
            for index, remote_path in enumerate(paths):
                if cancel_signal is not None and cancel_signal.is_set():
                    raise asyncio.CancelledError
                if not isinstance(remote_path, str) or not remote_path.startswith("/") or "\x00" in remote_path:
                    logger.warning("Skipping invalid printer file path: %r", remote_path)
                    failed_paths.append(remote_path)
                    continue
                staged_path = bundle_dir / f"download-{index}"
                try:
                    expected_size = sizes.get(remote_path)
                    if expected_size is not None:
                        free = (await asyncio.to_thread(shutil.disk_usage, root)).free
                        if free < expected_size + PRINTER_ZIP_FREE_SPACE_RESERVE:
                            raise PrinterFilesZipInsufficientSpaceError(
                                "The app data volume lacks space for the next selected file"
                            )
                    downloaded = await download_file_async(
                        printer.ip_address,
                        printer.access_code,
                        remote_path,
                        staged_path,
                        timeout=600,
                        socket_timeout=60,
                        printer_model=printer.model,
                        expected_size=expected_size,
                        max_bytes=MAX_PRINTER_ZIP_BYTES - total_bytes,
                        cancel_event=cancel_signal,
                        min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
                        # Outside the per-printer download gate (#2957), in both
                        # directions. A selection of ~250 MB /ipcam chunks holds
                        # the printer for as long as it legitimately takes, and
                        # nothing else should be made to wait that out; equally,
                        # each file here must not stall behind a thumbnail.
                        serialize=False,
                    )
                    if not downloaded:
                        failed_paths.append(remote_path)
                        continue
                    # Deliberately no second size comparison here. The transfer
                    # was already checked against the printer's own SIZE, which
                    # download_to_file treats as the authority precisely because
                    # it beats a hint the browser round-tripped; re-judging the
                    # result against that hint would overrule the better number
                    # with the worse one. The hint goes stale in exactly the case
                    # this feature exists for -- an /ipcam chunk or a timelapse
                    # still being written when the listing was taken -- and a
                    # complete file would then be dropped as "truncated".
                    file_size = (await asyncio.to_thread(staged_path.stat)).st_size
                    if total_bytes + file_size > MAX_PRINTER_ZIP_BYTES:
                        raise PrinterFilesZipTooLargeError(
                            f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit"
                        )
                    free = (await asyncio.to_thread(shutil.disk_usage, root)).free
                    if free < file_size + PRINTER_ZIP_FREE_SPACE_RESERVE:
                        raise PrinterFilesZipInsufficientSpaceError(
                            "The app data volume ran out of safe staging space while building the ZIP"
                        )
                    compression = (
                        zipfile.ZIP_STORED if remote_path.lower().endswith(VIDEO_SUFFIXES) else zipfile.ZIP_DEFLATED
                    )
                    arc_source = remote_path if preserve_paths else PurePosixPath(remote_path).name
                    await asyncio.to_thread(
                        archive.write,
                        staged_path,
                        _zip_arcname(arc_source, used_names),
                        compress_type=compression,
                    )
                    successful += 1
                    total_bytes += file_size
                except DownloadLimitExceeded as exc:
                    raise PrinterFilesZipTooLargeError(
                        f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit"
                    ) from exc
                except DownloadInsufficientSpace as exc:
                    raise PrinterFilesZipInsufficientSpaceError(
                        "The app data volume ran out of safe staging space during transfer"
                    ) from exc
                except DownloadCancelled as exc:
                    raise asyncio.CancelledError from exc
                except (PrinterFilesZipTooLargeError, PrinterFilesZipInsufficientSpaceError):
                    raise
                except Exception as exc:
                    logger.warning("Failed to add %s to printer ZIP: %s", remote_path, exc)
                    failed_paths.append(remote_path)
                finally:
                    await asyncio.to_thread(staged_path.unlink, missing_ok=True)
                    if progress_callback is not None:
                        await progress_callback(successful, len(failed_paths))
        finally:
            await asyncio.shield(asyncio.to_thread(archive.close))
    except BaseException:
        if bundle_dir is not None:
            await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
        raise

    if successful == 0 and not allow_empty:
        await asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)
        raise FileNotFoundError("No files could be downloaded")
    return PrinterFilesZipResult(
        path=zip_path,
        requested=len(paths),
        successful=successful,
        failed_paths=tuple(failed_paths),
        total_bytes=total_bytes,
    )


def printer_file_path(printer_id: int, token: str) -> Path | None:
    """Resolve a prepared native single-file download."""

    bundle_key = f"{printer_id}-{token}"
    if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
        return None
    return _printer_zip_root() / bundle_key / "printer-file"


def bind_printer_file_to_token(result: PrinterFilesZipResult, printer_id: int, token: str) -> PrinterFilesZipResult:
    target = printer_file_path(printer_id, token)
    if target is None:
        raise ValueError("Invalid printer file token")
    result.path.parent.rename(target.parent)
    return replace(result, path=target)


async def build_printer_file(
    printer,
    remote_path: str,
    expected_size: int | None,
    *,
    bundle_key: str,
    cancel_signal: _FileCancelSignal | None = None,
) -> PrinterFilesZipResult:
    """Stage one printer file on disk for a browser-native download.

    Also the read path for the 3MF preview in the file browser, which is why
    nothing here waits on a shared lock: a preview must not queue behind
    somebody else's ten-gigabyte selection for as long as that takes.
    """

    if not remote_path.startswith("/") or "\x00" in remote_path:
        raise FileNotFoundError("Invalid printer file path")
    root = await asyncio.to_thread(_ensure_printer_zip_root)
    size_hints = {remote_path: expected_size} if expected_size is not None else {}
    await asyncio.to_thread(_check_initial_space, root, size_hints)
    bundle_dir = root / bundle_key
    try:
        await asyncio.to_thread(bundle_dir.mkdir, mode=0o700)
        local_path = bundle_dir / "printer-file"
        downloaded = await download_file_async(
            printer.ip_address,
            printer.access_code,
            remote_path,
            local_path,
            timeout=600,
            socket_timeout=60,
            printer_model=printer.model,
            expected_size=expected_size,
            max_bytes=MAX_PRINTER_ZIP_BYTES,
            cancel_event=cancel_signal,
            min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
            # The lock-free promise in this function's docstring, kept: a preview
            # must not queue behind somebody else's selection (#2957).
            serialize=False,
        )
        if not downloaded:
            raise FileNotFoundError("The selected printer file could not be downloaded")
        file_size = (await asyncio.to_thread(local_path.stat)).st_size
        return PrinterFilesZipResult(
            path=local_path,
            requested=1,
            successful=1,
            failed_paths=(),
            total_bytes=file_size,
        )
    except DownloadLimitExceeded as exc:
        await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
        raise PrinterFilesZipTooLargeError(f"Downloaded file exceeds the {MAX_PRINTER_ZIP_BYTES}-byte limit") from exc
    except DownloadInsufficientSpace as exc:
        await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
        raise PrinterFilesZipInsufficientSpaceError(
            "The app data volume ran out of safe staging space during transfer"
        ) from exc
    except DownloadCancelled as exc:
        await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
        raise asyncio.CancelledError from exc
    except BaseException:
        await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
        raise


async def _run_printer_files_job(
    printer,
    job_id: str,
    paths: list[str],
    sizes: dict[str, int],
    filename: str,
    as_zip: bool,
) -> None:
    from backend.app.core.auth import create_slicer_download_token

    cancel_signal = _FileCancelSignal(_job_cancel_path(job_id))
    status = PrinterFilesJobStatus(job_id, printer.id, "preparing", len(paths), filename=filename)
    await asyncio.to_thread(_write_job_status, status)

    async def report_progress(successful: int, failed: int) -> None:
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(
                job_id,
                printer.id,
                "preparing",
                len(paths),
                successful=successful,
                failed=failed,
                filename=filename,
            ),
        )

    try:
        async with asyncio.timeout(MAX_PRINTER_ZIP_PREPARE_SECONDS):
            if as_zip:
                result = await build_printer_files_zip(
                    printer,
                    paths,
                    sizes,
                    bundle_key=f"job-{job_id}",
                    cancel_signal=cancel_signal,
                    progress_callback=report_progress,
                )
            else:
                result = await build_printer_file(
                    printer,
                    paths[0],
                    sizes.get(paths[0]),
                    bundle_key=f"job-{job_id}",
                    cancel_signal=cancel_signal,
                )
        if cancel_signal.is_set():
            await asyncio.to_thread(remove_printer_files_zip, result.path)
            raise asyncio.CancelledError
        token = await create_slicer_download_token("printer-files", printer.id)
        if as_zip:
            result = await asyncio.to_thread(bind_printer_files_zip_to_token, result, printer.id, token)
        else:
            result = await asyncio.to_thread(bind_printer_file_to_token, result, printer.id, token)
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(
                job_id,
                printer.id,
                "ready",
                len(paths),
                successful=result.successful,
                failed=len(result.failed_paths),
                token=token,
                filename=filename,
            ),
        )
    except asyncio.CancelledError:
        await asyncio.shield(
            asyncio.to_thread(
                _write_job_status,
                PrinterFilesJobStatus(job_id, printer.id, "cancelled", len(paths), filename=filename),
            )
        )
    except PrinterFilesZipTooLargeError as exc:
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
        )
    except PrinterFilesZipInsufficientSpaceError as exc:
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
        )
    except TimeoutError:
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(
                job_id,
                printer.id,
                "failed",
                len(paths),
                filename=filename,
                message="Printer download preparation exceeded the 30-minute limit",
            ),
        )
    except FileNotFoundError as exc:
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
        )
    except Exception:
        logger.exception("Printer download job %s failed", job_id)
        await asyncio.to_thread(
            _write_job_status,
            PrinterFilesJobStatus(
                job_id,
                printer.id,
                "failed",
                len(paths),
                filename=filename,
                message="Printer download preparation failed",
            ),
        )
    finally:
        await asyncio.to_thread(_job_cancel_path(job_id).unlink, missing_ok=True)


async def start_printer_files_job(
    printer,
    paths: list[str],
    sizes: dict[str, int],
    filename: str,
    *,
    as_zip: bool,
) -> PrinterFilesJobStatus:
    """Start a bounded background preparation and return immediately."""

    if not paths:
        raise ValueError("No files specified")
    root = await asyncio.to_thread(_ensure_printer_zip_root)
    await asyncio.to_thread(_prune_stale_bundles, root)
    await asyncio.to_thread(_check_initial_space, root, sizes)
    job_id = secrets.token_urlsafe(24)
    status = PrinterFilesJobStatus(job_id, printer.id, "queued", len(paths), filename=filename)
    await asyncio.to_thread(_write_job_status, status)
    task = spawn_background_task(
        _run_printer_files_job(printer, job_id, paths, sizes, filename, as_zip),
        name=f"printer-download-{printer.id}-{job_id}",
    )
    _LOCAL_JOB_TASKS[job_id] = task
    task.add_done_callback(lambda _task: _LOCAL_JOB_TASKS.pop(job_id, None))
    return status


async def get_printer_files_job(job_id: str, printer_id: int) -> PrinterFilesJobStatus | None:
    status = await asyncio.to_thread(_read_job_status, job_id)
    if status is None or status.printer_id != printer_id:
        return None
    return status


async def cancel_printer_files_job(job_id: str, printer_id: int) -> bool:
    status = await get_printer_files_job(job_id, printer_id)
    if status is None:
        return False
    await asyncio.to_thread(_job_cancel_path(job_id).touch)
    task = _LOCAL_JOB_TASKS.get(job_id)
    if task is not None and not task.done():
        task.cancel()
    if status.state == "ready" and status.token:
        zip_path = printer_files_zip_path(printer_id, status.token)
        prepared = (
            zip_path
            if zip_path is not None and await asyncio.to_thread(zip_path.is_file)
            else printer_file_path(printer_id, status.token)
        )
        if prepared is not None:
            await asyncio.to_thread(remove_printer_files_zip, prepared)
        await asyncio.to_thread(
            _write_job_status,
            replace(status, state="cancelled", token=None),
        )
    return True


def remove_printer_files_zip(zip_path: Path) -> None:
    """Remove a completed download bundle after FileResponse finishes."""

    shutil.rmtree(zip_path.parent, ignore_errors=True)