Re-run search whenever file list updates.

- Successful SSO /auth/api/validate responses are cached per credential
  and perm/renew URL for 10s, so watch websocket re-checks do not hammer
  the auth backend. A POST to the logout endpoint purges all cached
  entries for the request's credentials immediately, so logout/login
  flows are not served stale successes.
- The watch websocket now re-validates auth before each forwarded
  message and every 10s when idle (SSO and built-in sessions alike).
  When the session is gone the client gets an auth error message and
  the socket is closed, instead of streaming updates forever.
- Token-authenticated (API/share token) sockets are exempt from
  re-validation; they are checked once at handshake.

cista/api.py

@@ -2,11 +2,11 @@ import asyncio
 from secrets import token_bytes
 
 import msgspec
+from mediapreview.office import is_available_cached
 from sanic import Blueprint, json
 from sanic.exceptions import BadRequest
-from sanic.log import logger
 
-from cista import __version__, auth, config, onlyoffice, sharefs, sso, watching
+from cista import __version__, auth, config, sharefs, sso, watching
 from cista.auth import (
     create_share_token_handler,
     create_token_handler,
@@ -14,7 +14,11 @@ from cista.auth import (
     list_tokens_handler,
 )
 from cista.fileio import FileServer
-from cista.util.apphelpers import websocket_wrapper
+from cista.util.apphelpers import (
+    get_watch_user_info,
+    run_auth_checked_watch,
+    websocket_wrapper,
+)
 
 bp = Blueprint("api", url_prefix="/api")
 fileserver = FileServer()
@@ -35,29 +39,7 @@ async def stop_fileserver(app):
 @bp.websocket("watch")
 @websocket_wrapper
 async def watch(req, ws):
-    # Build user info from either built-in auth or SSO
-    user_info = None
-    if sso.paskia_enabled():
-        # SSO auth: call validation to get user info (don't enforce auth in public mode)
-        try:
-            # WebSocket cannot forward Set-Cookie, so ask the auth backend not to
-            # renew the session here; renewal happens on the HTTP side instead.
-            await sso.validate_sso_request(req, renew=False)
-        except Exception as e:
-            logger.debug("watch SSO validation failed: %s", e)
-        if sso_user := getattr(req.ctx, "sso_user", None):
-            ctx = sso_user.get("ctx", {})
-            perms = ctx.get("permissions", [])
-            user_info = {
-                "username": ctx.get("user", {}).get("display_name", ""),
-                "privileged": "cista:admin" in perms,
-            }
-    elif req.ctx.user:
-        # Built-in auth: use local user database
-        user_info = {
-            "username": req.ctx.username,
-            "privileged": req.ctx.user.privileged,
-        }
+    user_info = await get_watch_user_info(req)
 
     await ws.send(
         msgspec.json.encode(
@@ -67,7 +49,7 @@ async def watch(req, ws):
                     "version": __version__,
                     "public": config.config.public,
                     "paskia": sso.paskia_enabled(),
-                    "office_previews": await onlyoffice.is_available_cached(),
+                    "office_previews": await is_available_cached(),
                 },
                 "user": user_info,
             }
@@ -84,17 +66,7 @@ async def watch(req, ws):
             await ws.send(root)
         else:
             await ws.send(watching.format_root(sharefs.build_virtual_root(share_token)))
-        # Send updates
-        while True:
-            msg = await q.get()
-            if share_token is None or (
-                isinstance(msg, str) and msg.startswith('{"space"')
-            ):
-                await ws.send(msg)
-            else:
-                await ws.send(
-                    watching.format_root(sharefs.build_virtual_root(share_token))
-                )
+        await run_auth_checked_watch(req, ws, q, share_token)
     except RuntimeError as e:
         if str(e) == "cannot schedule new futures after shutdown":
             return  # Server shutting down, drop the WebSocket

cista/app.py

@@ -10,8 +10,10 @@ from wsgiref.handlers import format_date_time
 
 import tracerite
 from blake3 import blake3
+from mediapreview.office import close_oo_client, log_reachable_info
+from mediapreview.pool import shutdown_preview_workers, start_preview_workers
 from sanic import Sanic, empty, raw, redirect
-from sanic.exceptions import Forbidden, NotFound
+from sanic.exceptions import Forbidden, NotFound, RequestCancelled
 from sanic.log import logger
 from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
@@ -29,11 +31,11 @@ from cista import (
     watching,
 )
 from cista.api import bp
-from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.sanic_logging import (
     configure_access_logging,
     configure_main_logging,
     format_access_log,
+    reset_sanic_log_levels,
 )
 from cista.sanic_logging import logger as access_logger
 from cista.util.apphelpers import handle_sanic_exception
@@ -100,6 +102,19 @@ async def forward_sso_cookies(req, res):
             res.headers.add("set-cookie", cookie)
 
 
+@app.on_response
+async def invalidate_sso_cache_on_logout(req, _res):
+    """Purge cached SSO validations after a logout request."""
+    # Convenience for logout/login flows, not a security feature: cached
+    # entries expire after 10 seconds anyway if the logout happened elsewhere.
+    if (
+        sso.paskia_enabled()
+        and req.method == "POST"
+        and req.path in {"/auth/api/logout", "/auth/logout"}
+    ):
+        sso.invalidate_validation_cache(req)
+
+
 @app.on_response
 async def persist_auth_session(req, res):
     """Persist a session cookie after successful Authorization-based auth."""
@@ -123,12 +138,30 @@ app.blueprint(fileserver.bp)
 app.exception(Exception)(handle_sanic_exception)
 
 
+@app.exception(asyncio.CancelledError)
+async def request_cancelled(req, e):
+    """Request cancelled mid-flight (client disconnect or server shutdown).
+
+    Sanic wraps this as RequestCancelled (client disconnect only) — a
+    BaseException, so the generic Exception handler above never sees it — and
+    its default handler renders a 500 error page. Report 499 for client
+    disconnects and 503 for server-side cancellation instead; no traceback
+    (quiet=True), since there is nothing to fix. The connection is usually
+    already gone.
+    """
+    if not getattr(req.ctx, "log_extra", None):
+        req.ctx.log_extra = "cancelled"
+    return empty(499 if isinstance(e, RequestCancelled) else 503)
+
+
 setproctitle("cista-main")
 
 
 @app.before_server_start
 async def main_start(app):
+    reset_sanic_log_levels()
     config.load_config()
+    onlyoffice.configure()
     setproctitle(f"cista {config.config.path.name}")
     app.ctx.threadexec = ThreadPoolExecutor(
         max_workers=4, thread_name_prefix="cista-worker"
@@ -142,7 +175,7 @@ async def main_start(app):
 @app.after_server_start
 async def main_after_start(app):
     _ = app
-    onlyoffice.log_reachable_info()
+    log_reachable_info()
 
 
 # Sanic sometimes fails to execute after_server_stop, so we do it before instead (potentially interrupting handlers)
@@ -150,7 +183,7 @@ async def main_after_start(app):
 async def main_stop(app):
     async with asyncio.TaskGroup() as tg:
         tg.create_task(asyncio.to_thread(watching.stop, app))
-        tg.create_task(onlyoffice.close_oo_client())
+        tg.create_task(close_oo_client())
         tg.create_task(shutdown_preview_workers())
         tg.create_task(sso.close_client())
 

cista/docker/Dockerfile

@@ -1,70 +0,0 @@
-# Patched OnlyOffice Document Server with configurable converter worker count.
-#
-# The Community Edition hardcodes the document converter to 1 worker,
-# which creates a severe bottleneck under concurrent load.
-# This image patches the open-source license.js to spawn a configurable
-# number of converter workers (default 8).
-#
-# Build:
-#   docker build -t onlyoffice-cista docker/onlyoffice-converter-patch
-#
-# Run:
-#   docker run -d -p 8988:80 \
-#     -e WORKERS=16 \
-#     -e JWT_SECRET=your-strong-secret \
-#     --name onlyoffice onlyoffice-cista
-#
-# JWT:
-#   Set JWT_SECRET to the same value you pass to Cista as ONLYOFFICE_JWT_SECRET.
-#   OnlyOffice will enable token validation automatically.
-#
-# The ONLYOFFICE_VERSION build arg lets you target a specific release.
-
-ARG ONLYOFFICE_VERSION=9.3.1
-
-FROM onlyoffice/documentserver:${ONLYOFFICE_VERSION}
-
-# Prevent interactive apt prompts
-ENV DEBIAN_FRONTEND=noninteractive
-
-# Install Node.js, npm, and git so we can run the FileConverter from source.
-RUN apt-get update -qq && \
-    apt-get install -y -qq --no-install-recommends \
-        nodejs \
-        npm \
-        git \
-        ca-certificates && \
-    rm -rf /var/lib/apt/lists/*
-
-# Clone the open-source server components (shallow, ~15 MB).
-# The master branch is used because the Linux/web tags are not published
-# in the server repo; the license.js file has been stable for years.
-RUN git clone --depth 1 https://github.com/ONLYOFFICE/server.git /opt/oo-server
-
-# Patch license.js so the converter worker count is read from an env var
-# instead of being hardcoded to 1.
-RUN sed -i \
-    's/count: 1,/count: parseInt(process.env.WORKERS, 10) || 8,/' \
-    /opt/oo-server/Common/sources/license.js
-
-# Install npm dependencies for the modules the FileConverter touches.
-# DocService deps are also needed because converter.js pulls in baseConnector.
-RUN cd /opt/oo-server/Common      && npm ci --no-audit --no-fund
-RUN cd /opt/oo-server/FileConverter && npm ci --no-audit --no-fund
-RUN cd /opt/oo-server/DocService  && npm ci --no-audit --no-fund
-
-# Back up the compiled pkg binary and replace it with our wrapper.
-RUN mv /var/www/onlyoffice/documentserver/server/FileConverter/converter \
-       /var/www/onlyoffice/documentserver/server/FileConverter/converter.orig
-
-COPY converter-wrapper.sh /var/www/onlyoffice/documentserver/server/FileConverter/converter
-RUN chmod +x /var/www/onlyoffice/documentserver/server/FileConverter/converter
-
-# Default worker count (override at runtime with -e WORKERS=16).
-ENV WORKERS=8
-
-# Use our custom entrypoint to persist the env var to a file that the
-# non-root converter process (user=ds) can read.
-COPY entrypoint.sh /app/ds/run-document-server-patched.sh
-RUN chmod +x /app/ds/run-document-server-patched.sh
-ENTRYPOINT ["/app/ds/run-document-server-patched.sh"]

cista/docker/converter-wrapper.sh

@@ -1,19 +0,0 @@
-#!/bin/bash
-# Wrapper that runs the OnlyOffice FileConverter from patched Node.js source.
-# Replaces the compiled pkg binary shipped with the Community Edition.
-
-# The env var is not passed through supervisor to the 'ds' user, so we read
-# it from a file written by the custom entrypoint.
-if [ -z "${WORKERS}" ] && [ -r /tmp/oo-converter-workers.txt ]; then
-  export WORKERS=$(cat /tmp/oo-converter-workers.txt)
-fi
-
-cd /opt/oo-server/FileConverter || exit 1
-
-export NODE_ENV=production-linux
-export NODE_CONFIG_DIR=/etc/onlyoffice/documentserver
-export NODE_DISABLE_COLORS=1
-export APPLICATION_NAME=onlyoffice
-export LD_LIBRARY_PATH=/var/www/onlyoffice/documentserver/server/FileConverter/bin
-
-exec node sources/convertermaster.js "$@"

cista/docker/entrypoint.sh

@@ -1,8 +0,0 @@
-#!/bin/bash
-# Custom entrypoint that persists WORKERS to a file readable by
-# the non-root user that supervisor uses to run the converter.
-
-echo "${WORKERS:-8}" > /tmp/oo-converter-workers.txt
-chmod 644 /tmp/oo-converter-workers.txt
-
-exec /app/ds/run-document-server.sh "$@"

cista/onlyoffice.py

@@ -1,134 +1,27 @@
-"""OnlyOffice Document Server integration for office document preview.
+"""Cista-specific OnlyOffice setup.
 
-Provides server-side conversion of office documents to PNG via the
-OnlyOffice Document Server /ConvertService.ashx API. The resulting PNG
-is passed through pyvips for AVIF compression.
-
-Environment requirements:
-    - OnlyOffice Document Server must be running and reachable.
-    - If Document Server runs in Docker, the callback host IP must be
-      reachable from the container (usually the docker bridge IP).
+The conversion client itself lives in `mediapreview.office`; this module
+only bridges cista's config-derived JWT secret into it and wires the
+`--oosetup` Docker bootstrap to cista's config.
 """
 
-import asyncio
-import json
 import os
-import socket
-import socketserver
-import subprocess
-import threading
-import urllib.error
-import urllib.request
-from functools import partial
-from http.server import SimpleHTTPRequestHandler
 from pathlib import Path
-from time import perf_counter
-from urllib.parse import quote
 
-import httpx
-import jwt
-from sanic.log import logger
+import mediapreview.office
 
 from cista import config
 
-# ---------------------------------------------------------------------------
-# Configuration helpers
-# ---------------------------------------------------------------------------
 
-
-_httpx_client: httpx.AsyncClient | None = None
-
-
-def _get_onlyoffice_url() -> str:
-    return os.environ.get("ONLYOFFICE_CISTA_URL", "http://localhost:8988")
-
-
-def _get_jwt_secret() -> str:
-    return (
-        os.environ.get("ONLYOFFICE_JWT_SECRET")
-        or config.derived_secret("onlyoffice", size=16).hex()
+def configure() -> None:
+    """Point mediapreview's OnlyOffice client at cista's derived JWT secret."""
+    os.environ.setdefault(
+        "ONLYOFFICE_JWT_SECRET", config.derived_secret("onlyoffice", size=16).hex()
     )
 
 
-def _get_callback_host() -> str:
-    """Return the host IP that OnlyOffice (usually in Docker) can use to reach us."""
-    if host := os.environ.get("ONLYOFFICE_CALLBACK_HOST"):
-        return host
-    # Try to auto-detect docker bridge IP
-    try:
-        result = subprocess.run(
-            ["/sbin/ip", "-4", "addr", "show", "docker0"],
-            capture_output=True,
-            text=True,
-            timeout=2,
-            check=False,
-        )
-        for line in result.stdout.splitlines():
-            if "inet " in line:
-                parts = line.strip().split()
-                addr_part = parts[1]  # e.g. 172.17.0.1/16
-                return addr_part.split("/")[0]
-    except Exception:
-        logger.debug("Failed to auto-detect docker bridge IP")
-    return "127.0.0.1"
-
-
-# ---------------------------------------------------------------------------
-# Async HTTP client
-# ---------------------------------------------------------------------------
-
-
-def get_httpx_client() -> httpx.AsyncClient:
-    """Return the shared async HTTP client for OnlyOffice requests."""
-    global _httpx_client
-    if _httpx_client is None:
-        _httpx_client = httpx.AsyncClient()
-    return _httpx_client
-
-
-async def close_oo_client() -> None:
-    """Close the shared async HTTP client."""
-    global _httpx_client
-    if _httpx_client is not None:
-        await _httpx_client.aclose()
-        _httpx_client = None
-
-
-# ---------------------------------------------------------------------------
-# Availability check
-# ---------------------------------------------------------------------------
-
-
-def _probe_status() -> tuple[bool, bool, str | None]:
-    """Return (ok, responded, detail) for a lightweight reachability probe."""
-    url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx"
-    try:
-        with urllib.request.urlopen(url, timeout=2) as resp:  # noqa: S310
-            status = resp.status
-    except urllib.error.HTTPError as e:
-        status = e.code
-    except Exception:
-        return False, False, None
-
-    if status in (200, 405):
-        return True, True, None
-    if status >= 500:
-        return False, True, f"HTTP {status}"
-    return False, True, f"HTTP {status}"
-
-
-def log_reachable_info() -> None:
-    """Log info on success, warning on responded probe errors, silent on no-response."""
-    ok, responded, detail = _probe_status()
-    if ok:
-        logger.info("Using OnlyOffice document server at %s", _get_onlyoffice_url())
-    elif responded:
-        suffix = f": {detail}" if detail else ""
-        logger.warning("OnlyOffice probe failed%s", suffix)
-
-
-def setup_docker(confdir: Path | None = None) -> int:
-    """Build and run the patched OnlyOffice Docker image."""
+def setup_docker(confdir: Path | None = None) -> str:
+    """Build and run the patched OnlyOffice Docker image (via mediapreview)."""
     if confdir is not None:
         os.environ["CISTA_HOME"] = confdir.as_posix()
     config.init_confdir()
@@ -142,178 +35,5 @@ def setup_docker(confdir: Path | None = None) -> int:
                 "public": False,
             }
         )
-
-    secret = config.derived_secret("onlyoffice", size=16).hex()
-    docker_dir = Path(__file__).parent / "docker"
-    if not docker_dir.is_dir():
-        raise FileNotFoundError(
-            f"Docker files not found at {docker_dir}. Is the package installed correctly?"
-        )
-
-    logger.info("Building OnlyOffice image")
-    build_cmd = ["docker", "build", "-t", "onlyoffice-cista", str(docker_dir)]
-    logger.info("%s", " ".join(build_cmd))
-    result = subprocess.run(build_cmd, check=False, shell=False)  # noqa: S603
-    if result.returncode != 0:
-        raise RuntimeError("Failed to build OnlyOffice image")
-
-    logger.info("Starting OnlyOffice container")
-    run_cmd = [
-        "docker",
-        "run",
-        "-d",
-        "-p",
-        "8988:80",
-        "-e",
-        f"JWT_SECRET={secret}",
-        "-e",
-        "WORKERS=8",
-        "--name",
-        "onlyoffice-cista",
-        "--restart",
-        "unless-stopped",
-        "onlyoffice-cista",
-    ]
-    logger.info("%s", " ".join(run_cmd))
-    result = subprocess.run(run_cmd, check=False, shell=False)  # noqa: S603
-    if result.returncode != 0:
-        raise RuntimeError("Failed to start OnlyOffice container")
-    logger.info("OnlyOffice is running on http://localhost:8988")
-    return 0
-
-
-async def is_available_async(request_timeout: float = 2.0) -> bool:
-    """Return True if the configured OnlyOffice Document Server is reachable."""
-    url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx"
-    client = get_httpx_client()
-    try:
-        response = await client.get(url, timeout=request_timeout)
-    except Exception:
-        return False
-    else:
-        return response.status_code in (200, 405)
-
-
-_oo_available_cache: tuple[bool, float] | None = None
-OO_AVAILABILITY_CACHE_TTL = 30.0
-
-
-async def is_available_cached() -> bool:
-    """Return cached OnlyOffice availability, refreshed every 30 seconds."""
-    global _oo_available_cache
-    now = perf_counter()
-    if _oo_available_cache is not None:
-        result, timestamp = _oo_available_cache
-        if now - timestamp < OO_AVAILABILITY_CACHE_TTL:
-            return result
-    result = await is_available_async()
-    _oo_available_cache = (result, now)
-    return result
-
-
-# ---------------------------------------------------------------------------
-# Temporary HTTP server so OnlyOffice can download the file
-# ---------------------------------------------------------------------------
-
-
-class _QuietHandler(SimpleHTTPRequestHandler):
-    def log_message(self, fmt, *args) -> None:
-        pass
-
-
-def _get_free_port() -> int:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("0.0.0.0", 0))  # noqa: S104
-        return s.getsockname()[1]
-
-
-def _serve_file_temporarily(file_path: Path):
-    """Start a temporary HTTP server for *file_path* and return (url, server)."""
-    directory = str(file_path.parent)
-    filename = file_path.name
-    port = _get_free_port()
-
-    handler = partial(_QuietHandler, directory=directory)
-    httpd = socketserver.TCPServer(("0.0.0.0", port), handler)  # noqa: S104
-    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
-    thread.start()
-
-    host = _get_callback_host()
-    url = f"http://{host}:{port}/{quote(filename)}"
-    return url, httpd
-
-
-# ---------------------------------------------------------------------------
-# OnlyOffice conversion client
-# ---------------------------------------------------------------------------
-
-
-def _build_jwt_token(payload: dict) -> str | None:
-    secret = _get_jwt_secret()
-    if not secret:
-        return None
-    return jwt.encode(payload, secret, algorithm="HS256")
-
-
-async def convert_to_png_async(file_path: Path, request_timeout: float = 5.0) -> bytes:
-    """Convert *file_path* to PNG using OnlyOffice Document Server (async).
-
-    Returns the PNG bytes. Raises RuntimeError on failure.
-    """
-    oo_url = _get_onlyoffice_url().rstrip("/")
-    convert_url = f"{oo_url}/ConvertService.ashx"
-    client = get_httpx_client()
-
-    # Start temporary HTTP server so OnlyOffice can fetch the file
-    doc_url, httpd = await asyncio.to_thread(_serve_file_temporarily, file_path)
-    try:
-        suffix = file_path.suffix.lstrip(".").lower()
-        payload = {
-            "async": False,
-            "filetype": suffix,
-            "key": f"cista_{(await asyncio.to_thread(file_path.stat)).st_mtime_ns}",
-            "outputtype": "png",
-            "title": file_path.name,
-            "url": doc_url,
-        }
-
-        headers = {"Content-Type": "application/json"}
-        token = _build_jwt_token(payload)
-        if token:
-            # Conversion API expects JWT in request body when token checks are enabled.
-            payload["token"] = token
-            headers["Authorization"] = token
-
-        t_start = perf_counter()
-        response = await client.post(
-            convert_url,
-            content=json.dumps(payload).encode(),
-            headers=headers,
-            timeout=request_timeout,
-        )
-        response.raise_for_status()
-        body = response.content
-        t_end = perf_counter()
-
-        # Parse XML response
-        text = body.decode("utf-8", errors="replace")
-        if "<Error>" in text:
-            code = "unknown"
-            if "<Error>" in text and "</Error>" in text:
-                code = text.split("<Error>")[1].split("</Error>")[0]
-            raise RuntimeError(f"OnlyOffice conversion error: {code}")
-
-        if "<FileUrl>" not in text:
-            raise RuntimeError("OnlyOffice response did not contain FileUrl")
-
-        file_url = text.split("<FileUrl>")[1].split("</FileUrl>")[0]
-        file_url = file_url.replace("&amp;", "&")
-
-        logger.debug("OnlyOffice converted in %.2fs: %s", t_end - t_start, file_url)
-
-        # Download converted PNG
-        png_response = await client.get(file_url, timeout=request_timeout)
-        png_response.raise_for_status()
-        return png_response.content
-    finally:
-        await asyncio.to_thread(httpd.shutdown)
+    configure()
+    return mediapreview.office.setup_docker()

cista/preview.py

@@ -1,415 +1,40 @@
+"""Preview HTTP blueprint: routing, caching and response building.
+
+All conversion work is delegated to the mediapreview package (worker pool,
+OnlyOffice integration, classification); this module only wires it into
+Sanic with auth, etag negotiation and the in-memory response cache.
+"""
+
 import asyncio
-import contextlib
-import mimetypes
-import struct
-import sys
-import threading
 import urllib.parse
-from collections import OrderedDict
-from dataclasses import dataclass
-from multiprocessing import cpu_count
-from pathlib import Path, PurePosixPath
-from time import perf_counter
+from pathlib import PurePosixPath
 from urllib.parse import unquote
 from wsgiref.handlers import format_date_time
 
-import httpx
-import msgspec
-from blake3 import blake3
+from mediapreview import CachedPreview, PreviewCache, is_previewable_path
+from mediapreview.exceptions import (
+    PreviewBackendError,
+    PreviewCancelledError,
+    PreviewError,
+)
+from mediapreview.formats import OFFICE_PREVIEW_SUFFIXES
+from mediapreview.pool import (
+    generate_office_preview,
+    run_preview,
+)
 from sanic import Blueprint, empty, raw, redirect
 from sanic.exceptions import NotFound
 from sanic.log import logger
 
-from cista import auth, config, onlyoffice, sharefs, watching
+from cista import auth, config, sharefs, watching
 from cista.fileio import fuid
-from cista.preview_worker import (
-    DOC_PREVIEW_SUFFIXES,
-    OFFICE_PREVIEW_SUFFIXES,
-    PreviewRequest,
-    PreviewResponse,
-)
 from cista.util.filename import sanitize
 
 bp = Blueprint("preview", url_prefix="/preview")
 
-
-@dataclass(slots=True)
-class CachedPreview:
-    """Cached preview with headers and body."""
-
-    headers: dict[str, str]
-    body: bytes
-
-
-class PreviewCache:
-    """Thread-safe LRU cache for preview responses."""
-
-    def __init__(self, capacity: int = 500):
-        self.capacity = capacity
-        self._cache: OrderedDict[str, CachedPreview] = OrderedDict()
-        self._lock = threading.Lock()
-
-    def get(self, key: str) -> CachedPreview | None:
-        """Get cached preview, moving it to end (most recently used)."""
-        with self._lock:
-            if key in self._cache:
-                self._cache.move_to_end(key)
-                return self._cache[key]
-        return None
-
-    def set(self, key: str, value: CachedPreview) -> None:
-        """Cache preview, evicting oldest if at capacity."""
-        with self._lock:
-            if key in self._cache:
-                self._cache.move_to_end(key)
-            else:
-                if len(self._cache) >= self.capacity:
-                    self._cache.popitem(last=False)
-                self._cache[key] = value
-
-    def __len__(self) -> int:
-        with self._lock:
-            return len(self._cache)
-
-
 # Global preview cache instance
 _preview_cache = PreviewCache(capacity=500)
 
-PREVIEW_TIMEOUT = 10.0  # seconds until preview subprocess is killed
-PREVIEW_WORKERS = max(2, min(8, cpu_count()))
-_active_procs: set[asyncio.subprocess.Process] = set()
-_preview_pool = None
-_preview_pool_lock = asyncio.Lock()
-AVIF_FAST_EFFORT = 0
-WORKER_CHECKSUM_BYTES = 32
-WORKER_MAX_JSON_BYTES = 1_000_000
-
-
-class WorkerChecksumError(Exception):
-    """Raised when worker response checksum does not match the packet."""
-
-
-class WorkerProtocolError(Exception):
-    """Raised when worker response packet is malformed."""
-
-
-class _PreviewWorker:
-    def __init__(self, proc: asyncio.subprocess.Process):
-        self.proc = proc
-
-    async def request(
-        self,
-        filepath,
-        quality: int,
-        maxsize: int,
-        maxzoom: float,
-        data: bytes | None = None,
-    ):
-        if self.proc.returncode is not None:
-            raise WorkerProtocolError("worker already exited")
-        if self.proc.stdin is None or self.proc.stdout is None:
-            raise WorkerProtocolError("worker streams not available")
-
-        meta = msgspec.json.encode(
-            PreviewRequest(
-                path=str(filepath),
-                quality=quality,
-                maxsize=maxsize,
-                maxzoom=maxzoom,
-            )
-        )
-        payload = data or b""
-        packet = struct.pack("<II", len(meta), len(payload)) + meta + payload
-        self.proc.stdin.write(packet)
-        await self.proc.stdin.drain()
-
-        checksum = await self.proc.stdout.readexactly(WORKER_CHECKSUM_BYTES)
-        header = await self.proc.stdout.readexactly(8)
-        json_size, data_size = struct.unpack("<II", header)
-        if json_size > WORKER_MAX_JSON_BYTES:
-            raise WorkerProtocolError(f"worker JSON too large: {json_size}")
-        meta_raw = await self.proc.stdout.readexactly(json_size)
-        payload = await self.proc.stdout.readexactly(data_size)
-        packet = header + meta_raw + payload
-        if blake3(packet).digest() != checksum:
-            raise WorkerChecksumError("worker checksum mismatch")
-
-        resp = msgspec.json.decode(meta_raw, type=PreviewResponse)
-        if not resp.ok:
-            raise PreviewError(
-                resp.error or "preview worker error",
-                stderr=resp.stderr,
-                backend=resp.backend,
-            )
-        return payload or None, resp
-
-    async def kill(self) -> None:
-        if self.proc.returncode is None:
-            # Safe to hard-kill: the worker is stateless per request, and its
-            # subprocesses (ffmpeg) use stdin=DEVNULL so they never hold the
-            # worker's pipes open — proc.wait() cannot hang on pipe EOF.
-            with contextlib.suppress(ProcessLookupError):
-                self.proc.kill()
-            await self.proc.wait()
-        _active_procs.discard(self.proc)
-
-
-class _PreviewWorkerPool:
-    def __init__(self, size: int):
-        self.size = size
-        self._idle: asyncio.Queue[_PreviewWorker] = asyncio.Queue()
-        self._pending: asyncio.PriorityQueue[tuple[int, int, asyncio.Future, tuple]] = (
-            asyncio.PriorityQueue()
-        )
-        self._workers: set[_PreviewWorker] = set()
-        self._dispatchers: list[asyncio.Task] = []
-        self._seq = 0
-        self._closed = False
-
-    async def _read_startup_stderr(self, proc: asyncio.subprocess.Process) -> str:
-        if proc.stderr is None:
-            return ""
-        with contextlib.suppress(TimeoutError):
-            data = await asyncio.wait_for(proc.stderr.read(), timeout=0.5)
-            return data.decode(errors="replace").strip()
-        return ""
-
-    async def _spawn_worker(self) -> _PreviewWorker:
-        proc = await asyncio.create_subprocess_exec(
-            sys.executable,
-            "-m",
-            "cista.preview_worker",
-            stdin=asyncio.subprocess.PIPE,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        _active_procs.add(proc)
-        try:
-            ready = await asyncio.wait_for(proc.stdout.readexactly(1), timeout=30.0)
-        except TimeoutError as err:
-            with contextlib.suppress(ProcessLookupError):
-                proc.kill()
-            with contextlib.suppress(Exception):
-                await proc.wait()
-            stderr = await self._read_startup_stderr(proc)
-            if stderr:
-                raise WorkerProtocolError(
-                    "preview worker failed to become ready: " + stderr.splitlines()[-1]
-                ) from err
-            raise WorkerProtocolError("preview worker failed to become ready") from err
-        except asyncio.IncompleteReadError as err:
-            stderr = await self._read_startup_stderr(proc)
-            if stderr:
-                raise WorkerProtocolError(
-                    "preview worker exited before signalling readiness: "
-                    + stderr.splitlines()[-1]
-                ) from err
-            raise WorkerProtocolError(
-                "preview worker exited before signalling readiness"
-            ) from err
-        if ready != b"\x01":
-            raise WorkerProtocolError(f"preview worker ready signal invalid: {ready!r}")
-        return _PreviewWorker(proc)
-
-    async def _add_worker(self) -> None:
-        worker = await self._spawn_worker()
-        self._workers.add(worker)
-        await self._idle.put(worker)
-
-    async def _replace_worker(self, worker: _PreviewWorker) -> None:
-        self._workers.discard(worker)
-        await worker.kill()
-        if self._closed:
-            return
-        try:
-            await self._add_worker()
-        except Exception:
-            logger.exception("Failed to replace preview worker")
-
-    async def _dispatch_loop(self) -> None:
-        while True:
-            try:
-                _priority, _seq, future, args = await self._pending.get()
-            except asyncio.CancelledError:
-                return
-
-            if future.cancelled():
-                continue
-
-            try:
-                worker = await asyncio.wait_for(
-                    self._idle.get(), timeout=PREVIEW_TIMEOUT
-                )
-            except TimeoutError:
-                logger.warning(
-                    "Preview worker unavailable (%ds) for %s",
-                    int(PREVIEW_TIMEOUT),
-                    args[0].name,
-                )
-                if not future.done():
-                    future.set_exception(
-                        PreviewTimeoutError(
-                            args[0].name,
-                            backend=_expected_preview_backend(args[0]),
-                        )
-                    )
-                continue
-
-            filepath = args[0]
-            replace = False
-            try:
-                out, resp = await asyncio.wait_for(
-                    worker.request(*args),
-                    timeout=PREVIEW_TIMEOUT,
-                )
-                if not future.done():
-                    future.set_result((out, resp))
-            except TimeoutError:
-                replace = True
-                if not future.done():
-                    future.set_exception(
-                        PreviewTimeoutError(
-                            filepath.name,
-                            backend=_expected_preview_backend(filepath),
-                        )
-                    )
-            except WorkerChecksumError:
-                replace = True
-                logger.error("Preview checksum mismatch for %s", filepath.name)
-                if not future.done():
-                    future.set_exception(
-                        PreviewError(f"worker checksum mismatch for {filepath.name}")
-                    )
-            except PreviewError as e:
-                if not future.done():
-                    future.set_exception(e)
-            except (
-                WorkerProtocolError,
-                asyncio.IncompleteReadError,
-                BrokenPipeError,
-                ConnectionResetError,
-                OSError,
-                ValueError,
-                msgspec.json.DecodeError,
-            ) as e:
-                replace = True
-                logger.warning(
-                    "Preview worker protocol failure for %s: %s", filepath.name, e
-                )
-                if not future.done():
-                    future.set_exception(
-                        PreviewError(
-                            f"worker protocol failure for {filepath.name}: {e}"
-                        )
-                    )
-            except Exception:
-                replace = True
-                logger.exception(
-                    "Unexpected preview worker error for %s", filepath.name
-                )
-                if not future.done():
-                    future.set_exception(
-                        PreviewError(f"unexpected worker error for {filepath.name}")
-                    )
-            finally:
-                if replace:
-                    await self._replace_worker(worker)
-                elif worker.proc.returncode is None:
-                    await self._idle.put(worker)
-                else:
-                    await self._replace_worker(worker)
-
-    async def start(self) -> None:
-        workers = await asyncio.gather(
-            *(self._spawn_worker() for _ in range(self.size))
-        )
-        for worker in workers:
-            self._workers.add(worker)
-            await self._idle.put(worker)
-        for _ in range(self.size):
-            self._dispatchers.append(asyncio.create_task(self._dispatch_loop()))
-
-    async def run(
-        self,
-        filepath,
-        quality: int,
-        maxsize: int,
-        maxzoom: float,
-        data: bytes | None = None,
-    ):
-        if self._closed:
-            raise PreviewError("preview worker pool closed")
-        loop = asyncio.get_running_loop()
-        future = loop.create_future()
-        self._seq += 1
-        await self._pending.put(
-            (
-                _preview_job_priority(filepath),
-                self._seq,
-                future,
-                (filepath, quality, maxsize, maxzoom, data),
-            )
-        )
-        return await future
-
-    async def close(self) -> None:
-        self._closed = True
-        for task in self._dispatchers:
-            task.cancel()
-        if self._dispatchers:
-            await asyncio.gather(*self._dispatchers, return_exceptions=True)
-        self._dispatchers.clear()
-        workers = list(self._workers)
-        self._workers.clear()
-        while not self._pending.empty():
-            try:
-                _priority, _seq, future, _args = self._pending.get_nowait()
-            except asyncio.QueueEmpty:
-                break
-            if not future.done():
-                future.set_exception(PreviewError("preview worker pool closed"))
-        while not self._idle.empty():
-            try:
-                self._idle.get_nowait()
-            except asyncio.QueueEmpty:
-                break
-        await asyncio.gather(
-            *(worker.kill() for worker in workers), return_exceptions=True
-        )
-
-
-async def start_preview_workers() -> None:
-    """Warm up persistent preview workers during server startup."""
-    global _preview_pool
-    if _preview_pool is not None:
-        return
-    async with _preview_pool_lock:
-        if _preview_pool is not None:
-            return
-        pool = _PreviewWorkerPool(PREVIEW_WORKERS)
-        await pool.start()
-        _preview_pool = pool
-        logger.info("Started %d persistent preview workers", PREVIEW_WORKERS)
-
-
-async def shutdown_preview_workers() -> None:
-    """Kill persistent preview workers (called during server shutdown)."""
-    global _preview_pool
-    async with _preview_pool_lock:
-        pool = _preview_pool
-        _preview_pool = None
-    if pool is not None:
-        await pool.close()
-    if not _active_procs:
-        return
-    for proc in list(_active_procs):
-        with contextlib.suppress(ProcessLookupError):
-            proc.kill()
-    await asyncio.gather(
-        *(proc.wait() for proc in list(_active_procs)), return_exceptions=True
-    )
-    _active_procs.clear()
-
 
 @bp.on_request
 async def verify_preview(request):
@@ -417,178 +42,6 @@ async def verify_preview(request):
     await auth.verify(request)
 
 
-class PreviewTimeoutError(Exception):
-    """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
-
-    def __init__(self, message: str, *, backend: str | None = None):
-        super().__init__(message)
-        self.backend = backend
-
-
-class PreviewError(Exception):
-    """Raised when the preview subprocess exits with a non-zero status."""
-
-    def __init__(
-        self,
-        message: str,
-        *,
-        stderr: str | None = None,
-        backend: str | None = None,
-    ):
-        super().__init__(message)
-        self.stderr = stderr
-        self.backend = backend
-
-
-# Max concurrent OnlyOffice conversion requests. OO has its own queue;
-# we must not flood it. This is intentionally small.
-OO_MAX_CONCURRENT = PREVIEW_WORKERS
-
-
-class OOConversionManager:
-    """Manages async OnlyOffice conversions with deduplication and concurrency limits."""
-
-    def __init__(self, max_concurrent: int = OO_MAX_CONCURRENT):
-        self._semaphore = asyncio.Semaphore(max_concurrent)
-        self._in_flight: dict[str, asyncio.Future[bytes]] = {}
-        self._tasks: set[asyncio.Task[None]] = set()
-        self._lock = asyncio.Lock()
-
-    async def convert(self, filepath: Path) -> bytes:
-        """Return PNG bytes for *filepath*, deduplicating concurrent requests."""
-        stat = await asyncio.to_thread(filepath.stat)
-        key = f"{filepath}:{stat.st_mtime_ns}"
-
-        async with self._lock:
-            if key in self._in_flight:
-                future = self._in_flight[key]
-            else:
-                future = asyncio.get_running_loop().create_future()
-                self._in_flight[key] = future
-                task = asyncio.create_task(self._do_convert(filepath, key, future))
-                self._tasks.add(task)
-                task.add_done_callback(self._tasks.discard)
-
-        return await future
-
-    async def _do_convert(
-        self, filepath: Path, key: str, future: asyncio.Future[bytes]
-    ) -> None:
-        try:
-            async with self._semaphore:
-                png_bytes = await onlyoffice.convert_to_png_async(
-                    filepath, request_timeout=5.0
-                )
-        except Exception as e:
-            if not future.done():
-                future.set_exception(e)
-            async with self._lock:
-                self._in_flight.pop(key, None)
-        else:
-            if not future.done():
-                future.set_result(png_bytes)
-            async with self._lock:
-                self._in_flight.pop(key, None)
-
-
-_oo_manager: OOConversionManager | None = None
-
-
-def get_oo_manager() -> OOConversionManager:
-    """Return the singleton OOConversionManager."""
-    global _oo_manager
-    if _oo_manager is None:
-        _oo_manager = OOConversionManager(max_concurrent=OO_MAX_CONCURRENT)
-    return _oo_manager
-
-
-async def _generate_office_preview(
-    filepath: Path, quality: int, maxsize: int, maxzoom: float
-) -> tuple[bytes | None, PreviewResponse | None]:
-    """Generate a preview for an office file using OnlyOffice + worker AVIF conversion."""
-    manager = get_oo_manager()
-    t_oo_start = perf_counter()
-    png_bytes = await manager.convert(filepath)
-    t_oo_end = perf_counter()
-
-    img, resp = await _run_preview_process(
-        filepath, quality, maxsize, maxzoom, data=png_bytes
-    )
-
-    if resp is not None:
-        resp.backend = "onlyoffice+" + (resp.backend or "pyvips")
-        if resp.timings:
-            resp.timings = [round((t_oo_end - t_oo_start) * 1000, 1), *resp.timings]
-    return img, resp
-
-
-async def _run_preview_process(
-    filepath, quality: int, maxsize: int, maxzoom: float, data: bytes | None = None
-) -> tuple[bytes | None, PreviewResponse | None]:
-    """Run preview request in a persistent worker process."""
-    await start_preview_workers()
-    if _preview_pool is None:
-        raise PreviewError(f"preview worker pool unavailable for {filepath.name}")
-    return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data)
-
-
-def _onlyoffice_error_short_text(detail: str) -> str:
-    if detail.startswith("OnlyOffice conversion error:"):
-        code = detail.rsplit(":", 1)[-1].strip()
-        return {
-            "-8": "onlyoffice jwt error",
-            "-4": "onlyoffice input error",
-            "-2": "onlyoffice timeout error",
-            "-1": "onlyoffice unknown error",
-        }.get(code, f"onlyoffice {code} error")
-    if "OnlyOffice response did not contain FileUrl" in detail:
-        return "onlyoffice no-fileurl error"
-    return "onlyoffice error"
-
-
-def _preview_job_priority(path) -> int:
-    """Return priority for preview job (lower=higher priority).
-
-    Priority order: images (0) < video (1) < PDF (2) < office (3) < unknown (4)
-    """
-    suffix = path.suffix.lower()
-    if suffix in DOC_PREVIEW_SUFFIXES:
-        return 2
-    if suffix in OFFICE_PREVIEW_SUFFIXES:
-        return 3
-    mime_type, _ = mimetypes.guess_type(path.name)
-    if mime_type and mime_type.startswith("image/"):
-        return 0
-    if mime_type and mime_type.startswith("video/"):
-        return 1
-    return 4
-
-
-def _expected_preview_backend(path: Path) -> str:
-    """Best-effort backend label used for timeout/access logging."""
-    suffix = path.suffix.lower()
-    if suffix in OFFICE_PREVIEW_SUFFIXES:
-        return "onlyoffice"
-    if suffix in DOC_PREVIEW_SUFFIXES:
-        return "pdf"
-    mime_type, _ = mimetypes.guess_type(path.name)
-    if mime_type and mime_type.startswith("video/"):
-        return "video"
-    if mime_type and mime_type.startswith("image/"):
-        return "pyvips"
-    return "preview"
-
-
-def is_previewable_path(path) -> bool:
-    suffix = path.suffix.lower()
-    if suffix in DOC_PREVIEW_SUFFIXES or suffix in OFFICE_PREVIEW_SUFFIXES:
-        return True
-    mime_type, _ = mimetypes.guess_type(path.name)
-    if not mime_type:
-        return False
-    return mime_type.startswith(("image/", "video/"))
-
-
 @bp.get("/<path:path>")
 async def preview(req, path):
     """Preview a file"""
@@ -630,48 +83,27 @@ async def preview(req, path):
     # Generate preview
     try:
         if filepath.suffix.lower() in OFFICE_PREVIEW_SUFFIXES:
-            img, preview_resp = await asyncio.wait_for(
-                _generate_office_preview(filepath, quality, maxsize, maxzoom),
-                timeout=PREVIEW_TIMEOUT,
+            img, preview_resp = await generate_office_preview(
+                filepath, quality, maxsize, maxzoom
             )
         else:
-            img, preview_resp = await asyncio.wait_for(
-                _run_preview_process(filepath, quality, maxsize, maxzoom),
-                timeout=PREVIEW_TIMEOUT,
-            )
-    except TimeoutError:
-        req.ctx.log_extra = f"{_expected_preview_backend(filepath)} timeout"
-        return empty(503)
-    except PreviewTimeoutError as e:
-        req.ctx.log_extra = (
-            f"{(e.backend or _expected_preview_backend(filepath))} timeout"
-        )
-        return empty(503)
-    except httpx.HTTPStatusError:
-        req.ctx.log_extra = "onlyoffice N/A"
-        return empty(503)
-    except httpx.RequestError:
-        req.ctx.log_extra = "onlyoffice N/A"
-        return empty(503)
-    except RuntimeError as e:
-        detail = str(e)
-        if detail.startswith("OnlyOffice"):
-            req.ctx.log_extra = _onlyoffice_error_short_text(detail)
-            return empty(503)
-        raise
+            img, preview_resp = await run_preview(filepath, quality, maxsize, maxzoom)
     except PreviewError as e:
-        if e.backend:
-            req.ctx.log_extra = e.backend
-        detail = str(e)
-        if detail == "preview worker error" and e.stderr:
-            captured = e.stderr.strip()
-            if captured:
-                detail = captured.splitlines()[0]
-        logger.error("%s preview: %s", filepath, detail)
-        return empty(422)
+        # mediapreview is responsible for backend-specific diagnostics; cista only
+        # needs the backend name, a short access-log reason, and a response status.
+        if isinstance(e, PreviewCancelledError):
+            req.ctx.log_extra = e.short or "preview cancelled"
+            raise asyncio.CancelledError from e
+        status = 422 if isinstance(e, PreviewBackendError) else 503
+        req.ctx.log_extra = f"{e.backend}: {e.short}" if e.backend else e.short
+        if req.app.debug:
+            logger.warning("%s", str(e))
+        return empty(status)
     except asyncio.CancelledError:
+        # Server shutdown or client disconnect: the connection is being torn
+        # down, so responding is impossible — just annotate the access log.
         req.ctx.log_extra = "preview cancelled"
-        return empty(503)
+        raise
     except Exception:
         logger.exception("Unhandled preview error for %s", filepath)
         return empty(500)

cista/preview_worker.py

@@ -1,568 +0,0 @@
-"""Preview generation worker subprocess and synchronous preview engine.
-
-Two modes are supported:
-1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom.
-2) Long-lived mode: read framed requests from stdin and write framed responses.
-
-Framed request format (stdin):
-    (uint32 json size)(uint32 data size)(json)(binary data)
-
-Framed response format (stdout):
-    (blake3(packet))(uint32 json size)(uint32 payload size)(json)(binary payload)
-where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
-"""
-
-import contextlib
-import gc
-import io
-import logging
-import mimetypes
-import shlex
-import struct
-import subprocess
-import sys
-import tempfile
-from pathlib import Path
-from time import perf_counter
-
-import av
-import fitz  # PyMuPDF
-import msgspec
-import numpy as np
-import pyvips
-from blake3 import blake3
-
-from cista import config
-
-logger = logging.getLogger(__name__)
-
-AVIF_FAST_EFFORT = 0
-
-DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
-
-OFFICE_PREVIEW_SUFFIXES = {
-    ".doc",
-    ".dot",
-    ".docx",
-    ".docm",
-    ".dotx",
-    ".dotm",
-    ".rtf",
-    ".odt",
-    ".ott",
-    ".txt",
-    ".md",
-    ".mhtml",
-    ".mht",
-    ".html",
-    ".htm",
-    ".xml",
-    ".wps",
-    ".wri",
-    # Spreadsheets
-    ".xls",
-    ".xlsx",
-    ".xlsm",
-    ".xlsb",
-    ".xltx",
-    ".xltm",
-    ".ods",
-    ".ots",
-    ".csv",
-    # Presentations
-    ".ppt",
-    ".pptx",
-    ".pptm",
-    ".pps",
-    ".ppsx",
-    ".pot",
-    ".potx",
-    ".odp",
-    ".otp",
-}
-
-
-class PreviewRequest(msgspec.Struct, omit_defaults=True):
-    path: str
-    quality: int
-    maxsize: int
-    maxzoom: float
-
-
-class PreviewResponse(msgspec.Struct, omit_defaults=True):
-    ok: bool
-    mime: str | None = None
-    backend: str | None = None
-    timings: list[float] | None = None
-    error: str | None = None
-    stderr: str | None = None
-    width: int | None = None
-    height: int | None = None
-
-
-_enc = msgspec.json.Encoder()
-_dec_req = msgspec.json.Decoder(PreviewRequest)
-
-
-def _read_exactly(f, n: int) -> bytes:
-    buf = b""
-    while len(buf) < n:
-        chunk = f.read(n - len(buf))
-        if not chunk:
-            raise EOFError
-        buf += chunk
-    return buf
-
-
-def _read_request() -> tuple[PreviewRequest, bytes] | None:
-    try:
-        header = _read_exactly(sys.stdin.buffer, 8)
-    except EOFError:
-        return None
-    json_size, data_size = struct.unpack("<II", header)
-    meta_raw = _read_exactly(sys.stdin.buffer, json_size)
-    data = b""
-    if data_size:
-        data = _read_exactly(sys.stdin.buffer, data_size)
-    req = _dec_req.decode(meta_raw)
-    return req, data
-
-
-def _write_response(resp: PreviewResponse, payload: bytes) -> None:
-    meta_bytes = _enc.encode(resp)
-    packet = struct.pack("<II", len(meta_bytes), len(payload)) + meta_bytes + payload
-    checksum = blake3(packet).digest()
-    sys.stdout.buffer.write(checksum)
-    sys.stdout.buffer.write(packet)
-    sys.stdout.buffer.flush()
-
-
-def dispatch(path, quality, maxsize, maxzoom, data=None):
-    backend = "unknown"
-    try:
-        if data:
-            backend = "pyvips"
-            return process_image_buffer(
-                data, quality=quality, maxsize=maxsize, maxzoom=maxzoom
-            )
-        suffix = path.suffix.lower()
-        if suffix in DOC_PREVIEW_SUFFIXES:
-            backend = "pdf"
-            return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
-        mime_type, _ = mimetypes.guess_type(path.name)
-        if mime_type and mime_type.startswith("video/"):
-            backend = "video"
-            return process_video(path, quality=quality, maxsize=maxsize)
-        if mime_type and mime_type.startswith("image/"):
-            backend = "pyvips"
-            return process_image(path, quality=quality, maxsize=maxsize)
-    except ValueError as e:
-        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
-    except Exception as e:
-        logger.exception("Preview dispatch failed for %s", path)
-        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
-    return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")
-
-
-def process_image(path, *, maxsize, quality):
-    return process_image_pyvips(path, maxsize=maxsize, quality=quality)
-
-
-def _get_image_dimensions(path: Path) -> tuple[int, int] | None:
-    """Probe image dimensions.
-
-    pyvips can read the header of most formats (including HEIC) without
-    fully decoding the image.
-    """
-    try:
-        img = pyvips.Image.new_from_file(str(path))
-        img = img.autorot()
-    except pyvips.error.Error:
-        return None
-    else:
-        return img.width, img.height
-
-
-def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes:
-    """Convert any image to AVIF using ffmpeg CLI.
-
-    ffmpeg handles HEIC tile assembly, EXIF rotation, HDR metadata and
-    ICC profile embedding automatically.
-    """
-    dims = _get_image_dimensions(path)
-    crf = int(63 * (1 - quality / 100) ** 2)
-    with tempfile.NamedTemporaryFile(suffix=".avif", delete=False) as tmp_f:
-        tmp_path = tmp_f.name
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-i",
-        str(path),
-        "-frames:v",
-        "1",
-        "-c:v",
-        "av1",
-        "-crf",
-        str(crf),
-        "-cpu-used",
-        "8",
-        tmp_path,
-    ]
-    if dims is not None:
-        w, h = dims
-        if max(w, h) > maxsize:
-            scale = min(maxsize / w, maxsize / h)
-            new_w = int(w * scale)
-            new_h = int(h * scale)
-            # insert -s <wxh> right after the input file
-            cmd.insert(4, "-s")
-            cmd.insert(5, f"{new_w}x{new_h}")
-    try:
-        try:
-            # stdin=DEVNULL is critical: ffmpeg must not inherit the worker's
-            # stdin, which carries the framed request protocol. An inherited
-            # stdin lets ffmpeg eat protocol bytes and, if the worker is
-            # killed mid-conversion, keeps the orphaned ffmpeg holding the
-            # pipe open so the parent's proc.wait() hangs forever.
-            subprocess.run(  # noqa: S603
-                cmd,
-                capture_output=True,
-                check=True,
-                shell=False,
-                stdin=subprocess.DEVNULL,
-            )
-        except subprocess.CalledProcessError as e:
-            shell_cmd = shlex.join(cmd)
-            stderr = (e.stderr or b"").decode(errors="replace").strip()
-            if stderr:
-                raise RuntimeError(
-                    f"ffmpeg failed (exit {e.returncode}): {shell_cmd}\n{stderr}"
-                ) from e
-            raise RuntimeError(
-                f"ffmpeg failed (exit {e.returncode}): {shell_cmd}"
-            ) from e
-        with Path(tmp_path).open("rb") as f:
-            return f.read()
-    finally:
-        Path(tmp_path).unlink(missing_ok=True)
-
-
-def process_image_pyvips(path, *, maxsize, quality):
-    t_start = perf_counter()
-    suffix = path.suffix.lower()
-
-    # HEIC/HEIF: ffmpeg handles tile assembly and HDR correctly;
-    # skip pyvips entirely.
-    if suffix in (".heic", ".heif"):
-        heic_dims = _get_image_dimensions(path)
-        width, height = heic_dims or (None, None)
-        ret = _image_via_ffmpeg(path, maxsize, quality)
-        t_end = perf_counter()
-        return ret, PreviewResponse(
-            ok=True,
-            mime="image/avif",
-            backend="ffmpeg",
-            timings=[round((t_end - t_start) * 1000, 1)],
-            width=width,
-            height=height,
-        )
-
-    # Other image formats: pyvips first, ffmpeg fallback.
-    load_opts = {"access": "sequential"}
-    orig_w = orig_h = None
-    try:
-        img = pyvips.Image.new_from_file(str(path), **load_opts)
-        img = img.autorot()
-        orig_w, orig_h = img.width, img.height
-        scale = min(maxsize / img.width, maxsize / img.height, 1.0)
-        if scale < 1.0:
-            img = img.resize(scale)
-        ret = img.write_to_buffer(
-            ".avif",
-            Q=quality,
-            effort=AVIF_FAST_EFFORT,
-            strip=True,
-        )
-        backend = "pyvips"
-    except pyvips.error.Error:
-        orig_w, orig_h = None, None
-        ret = _image_via_ffmpeg(path, maxsize, quality)
-        backend = "ffmpeg"
-    t_end = perf_counter()
-
-    return ret, PreviewResponse(
-        ok=True,
-        mime="image/avif",
-        backend=backend,
-        timings=[round((t_end - t_start) * 1000, 1)],
-        width=orig_w,
-        height=orig_h,
-    )
-
-
-def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
-    _ = maxzoom
-    t_start = perf_counter()
-    img = pyvips.Image.new_from_buffer(data, "")
-    img = img.autorot()
-    orig_w, orig_h = img.width, img.height
-    scale = min(maxsize / img.width, maxsize / img.height, 1.0)
-    if scale < 1.0:
-        img = img.resize(scale)
-    ret = img.write_to_buffer(
-        ".avif",
-        Q=quality,
-        effort=AVIF_FAST_EFFORT,
-        strip=True,
-    )
-    t_end = perf_counter()
-
-    return ret, PreviewResponse(
-        ok=True,
-        mime="image/avif",
-        backend="pyvips",
-        timings=[round((t_end - t_start) * 1000, 1)],
-        width=orig_w,
-        height=orig_h,
-    )
-
-
-def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
-    t_load_start = perf_counter()
-    pdf = fitz.open(path)
-    page = pdf.load_page(page_number)
-    w, h = page.rect[2:4]
-    zoom = min(maxsize / w, maxsize / h, maxzoom)
-    mat = fitz.Matrix(zoom, zoom)
-    pix = page.get_pixmap(matrix=mat)
-    t_load_end = perf_counter()
-
-    t_save_start = perf_counter()
-    img = pyvips.Image.new_from_memory(
-        pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
-    )
-    ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True)
-    backend = "pdf+pyvips"
-    t_save_end = perf_counter()
-
-    return ret, PreviewResponse(
-        ok=True,
-        mime="image/avif",
-        backend=backend,
-        timings=[
-            round((t_load_end - t_load_start) * 1000, 1),
-            round((t_save_end - t_save_start) * 1000, 1),
-        ],
-        width=round(w),
-        height=round(h),
-    )
-
-
-def process_video(path, *, maxsize, quality):
-    frame = None
-    imgdata = io.BytesIO()
-    istream = ostream = icc = occ = frame = None
-    t_load_start = perf_counter()
-    # Initialize to avoid "possibly unbound" in static analysis when exceptions occur
-    t_load_end = t_load_start
-    t_save_start = t_load_start
-    t_save_end = t_load_start
-    with (
-        av.open(
-            str(path),
-            options={
-                "analyzeduration": "1000000",  # 1 second (in microseconds)
-                "fflags": "fastseek",
-            },
-        ) as icontainer,
-        av.open(imgdata, "w", format="avif") as ocontainer,
-    ):
-        istream = icontainer.streams.video[0]
-        istream.codec_context.skip_frame = "NONKEY"
-        icontainer.seek((icontainer.duration or 0) // 8)
-        for frame in icontainer.decode(istream):
-            if frame.dts is not None:
-                break
-        else:
-            raise RuntimeError("No frames found in video")
-
-        # Resize frame to thumbnail size
-        # Capture display dimensions before resize (accounting for rotation)
-        disp_w = frame.width
-        disp_h = frame.height
-        if frame.rotation in (90, 270):
-            disp_w, disp_h = disp_h, disp_w
-        if frame.width > maxsize or frame.height > maxsize:
-            scale_factor = min(maxsize / frame.width, maxsize / frame.height)
-            new_width = int(frame.width * scale_factor)
-            new_height = int(frame.height * scale_factor)
-            frame = frame.reformat(width=new_width, height=new_height)
-
-        # Apply EXIF rotation if present
-        if frame.rotation:
-            # frame.rotation indicates clockwise rotation needed to display correctly
-            # np.rot90 rotates counter-clockwise, so we negate k
-            k = (frame.rotation // 90) % 4  # Convert to counter-clockwise rotations
-            if k == 2:
-                # 180° rotation can be done in YUV420p, preserving HDR
-                try:
-                    fplanes = frame.to_ndarray()
-                    # Split into Y, U, V planes of proper dimensions
-                    planes = [
-                        fplanes[: frame.height],
-                        fplanes[
-                            frame.height : frame.height + frame.height // 4
-                        ].reshape(frame.height // 2, frame.width // 2),
-                        fplanes[frame.height + frame.height // 4 :].reshape(
-                            frame.height // 2, frame.width // 2
-                        ),
-                    ]
-                    # Rotate each plane by 180°
-                    planes = [np.rot90(p, 2) for p in planes]
-                    # Restore PyAV format
-                    planes = np.hstack([p.flat for p in planes]).reshape(
-                        -1, planes[0].shape[1]
-                    )
-                    frame = av.VideoFrame.from_ndarray(planes, format=frame.format.name)
-                    del planes, fplanes
-                except Exception:
-                    logger.exception("Error rotating video frame by 180°")
-            elif k in (1, 3):
-                # 90° or 270° rotation requires RGB conversion (loses HDR)
-                try:
-                    rgb = frame.to_ndarray(format="rgb24")
-                    rgb = np.rot90(rgb, k)
-                    frame = av.VideoFrame.from_ndarray(rgb, format="rgb24")
-                    frame = frame.reformat(
-                        format="yuv420p"
-                    )  # Convert back for encoding
-                    del rgb
-                except Exception:
-                    logger.exception(
-                        "Error rotating video frame by %s°", frame.rotation
-                    )
-
-        # libsvtav1 rejects full-range JPEG-style YUV pixel formats such as
-        # yuvj420p, so normalize them before opening the encoder.
-        if frame.format.name.startswith("yuvj"):
-            frame = frame.reformat(format="yuv420p")
-        t_load_end = perf_counter()
-
-        t_save_start = perf_counter()
-        crf = str(int(63 * (1 - quality / 100) ** 2))  # Closely matching PIL quality-%
-        ostream = ocontainer.add_stream(
-            "av1",
-            options={
-                "crf": crf,
-                "usage": "realtime",
-                "cpu-used": "8",
-                "threads": "1",
-            },
-        )
-        if not isinstance(ostream, av.VideoStream):
-            raise TypeError("failed to initialize AV1 video stream")
-        ostream.width = frame.width
-        ostream.height = frame.height
-        ostream.pix_fmt = frame.format.name
-        icc = istream.codec_context
-        occ = ostream.codec_context
-
-        # Copy HDR metadata from input video stream
-        occ.color_primaries = icc.color_primaries
-        occ.color_trc = icc.color_trc
-        occ.colorspace = icc.colorspace
-        occ.color_range = icc.color_range
-
-        ocontainer.mux(ostream.encode(frame))
-        ocontainer.mux(ostream.encode(None))  # Flush the stream
-        t_save_end = perf_counter()
-
-    # Capture result before cleanup
-    ret = imgdata.getvalue()
-    resp = PreviewResponse(
-        ok=True,
-        mime="image/avif",
-        backend="video",
-        timings=[
-            round((t_load_end - t_load_start) * 1000, 1),
-            round((t_save_end - t_save_start) * 1000, 1),
-        ],
-        width=disp_w,
-        height=disp_h,
-    )
-    del imgdata, istream, ostream, icc, occ, frame
-    gc.collect()
-    return ret, resp
-
-
-def _run_once() -> None:
-    if len(sys.argv) != 5:
-        sys.stderr.write(f"Usage: {sys.argv[0]} <path> <quality> <maxsize> <maxzoom>\n")
-        sys.exit(1)
-
-    path = Path(sys.argv[1])
-    quality = int(sys.argv[2])
-    maxsize = int(sys.argv[3])
-    maxzoom = float(sys.argv[4])
-    result, _ = dispatch(path, quality, maxsize, maxzoom)
-    if result:
-        sys.stdout.buffer.write(result)
-        sys.stdout.buffer.flush()
-
-
-def _run_loop() -> None:
-    while True:
-        result = _read_request()
-        if result is None:
-            return
-        req, data = result
-        stderr_capture = io.StringIO()
-        handler = logging.StreamHandler(stderr_capture)
-        root_logger = logging.getLogger()
-        root_logger.addHandler(handler)
-        try:
-            with contextlib.redirect_stderr(stderr_capture):
-                result, resp = dispatch(
-                    Path(req.path), req.quality, req.maxsize, req.maxzoom, data
-                )
-            if not resp.ok:
-                captured = stderr_capture.getvalue().strip()
-                if captured:
-                    resp = PreviewResponse(
-                        ok=False,
-                        backend=resp.backend,
-                        error=resp.error,
-                        stderr=captured,
-                    )
-            _write_response(resp, result or b"")
-        except Exception as e:
-            logger.exception("Preview worker error for %s", req.path)
-            captured = stderr_capture.getvalue().strip()
-            _write_response(
-                PreviewResponse(ok=False, error=str(e), stderr=captured or None), b""
-            )
-        finally:
-            root_logger.removeHandler(handler)
-            handler.close()
-
-
-def main() -> None:
-    # Configure all log output to stderr before any imports that may emit logs.
-    logging.basicConfig(stream=sys.stderr, level=logging.INFO)
-    try:
-        config.load_config()
-        logger.info("preview-worker config=%s", config.conffile)
-    except Exception:
-        logger.exception("preview-worker failed to load config at startup")
-    if len(sys.argv) > 1:
-        _run_once()
-        return
-    # Eagerly import heavy modules before signalling readiness so the parent
-    # does not hand us a request while we are still initialising.
-    sys.stdout.buffer.write(b"\x01")
-    sys.stdout.buffer.flush()
-    _run_loop()
-
-
-if __name__ == "__main__":
-    main()

cista/sanic_logging.py

@@ -3,11 +3,13 @@
 import logging
 import os
 import sys
-import unicodedata
 from ipaddress import IPv6Address
 
 from sanic.log import LOGGING_CONFIG_DEFAULTS
 
+from cista.util.logformat import EmojiFormatter as _EmojiFormatter
+from cista.util.logformat import display_width as _display_width
+
 logger = logging.getLogger("cista.access")
 
 
@@ -132,14 +134,6 @@ def format_duration_ms(duration_ms: float) -> str:
     return f"{hours}h{minutes}m"
 
 
-def _display_width(text: str) -> int:
-    return sum(
-        1 + (unicodedata.east_asian_width(c) in "FW")
-        for c in text
-        if unicodedata.category(c) != "Mn"
-    )
-
-
 def _format_left(label: str) -> str:
     return label[:19].ljust(19)
 
@@ -279,34 +273,45 @@ def configure_access_logging() -> None:
     logger.propagate = False
 
 
-_LEVEL_EMOJI = {
-    logging.DEBUG: "🔍",
-    logging.INFO: "ℹ️",  # noqa: RUF001
-    logging.WARNING: "⚠️",
-    logging.ERROR: "🛑",
-    logging.CRITICAL: "🛑",
-}
-
-
-def _format_level_prefix(levelno: int) -> str:
-    emoji = _LEVEL_EMOJI.get(levelno, "▪️")
-    prefix = f"{emoji} "
-    return prefix + (" " * max(0, 3 - _display_width(prefix)))
-
-
-class _EmojiFormatter(logging.Formatter):
-    """Compact formatter: emoji + message, no timestamp/level text/logger name."""
-
-    def format(self, record: logging.LogRecord) -> str:
-        return _format_level_prefix(record.levelno) + record.getMessage()
-
-
 def configure_main_logging() -> None:
-    """Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format.
+    """Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format
 
-    Patches LOGGING_CONFIG_DEFAULTS so the formatter survives every dictConfig
-    call Sanic makes during serve_single() / serve().
+    and make sure the root logger catches unhandled loggers instead of falling back
+    to logging.lastResort (which prints a bare message with no level prefix).
+
+    Patches LOGGING_CONFIG_DEFAULTS so the formatter and root logger survive every
+    dictConfig call Sanic makes during serve_single() / serve().
     """
+    # Give the root logger a real handler so third-party warnings (e.g.
+    # mediapreview.office) are formatted with the emoji prefix instead of being
+    # printed plain by logging.lastResort.
+    root = logging.getLogger()
+    root.setLevel(logging.WARNING)
+    if not root.handlers:
+        root_handler = ReentrantSafeStreamHandler(sys.stderr)
+        root_handler.setFormatter(_EmojiFormatter())
+        root.addHandler(root_handler)
+
+    # Ensure future dictConfig calls keep a root logger so unhandled loggers still
+    # get the emoji formatter rather than falling back to logging.lastResort.
+    LOGGING_CONFIG_DEFAULTS["root"] = {
+        "level": "WARNING",
+        "handlers": ["error_console"],
+    }
+
+    # Sanic's loggers already have their own handlers; stop them from bubbling up
+    # to the root handler we just added so messages are not duplicated.
+    for name in (
+        "sanic.root",
+        "sanic.error",
+        "sanic.access",
+        "sanic.server",
+        "sanic.websockets",
+    ):
+        logging.getLogger(name).propagate = False
+        if name in LOGGING_CONFIG_DEFAULTS["loggers"]:
+            LOGGING_CONFIG_DEFAULTS["loggers"][name]["propagate"] = False
+
     for handler_name in ("console", "error_console", "access_console"):
         LOGGING_CONFIG_DEFAULTS["handlers"][handler_name]["class"] = (
             "cista.sanic_logging.ReentrantSafeStreamHandler"
@@ -314,7 +319,30 @@ def configure_main_logging() -> None:
     LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
         "class": "cista.sanic_logging._EmojiFormatter",
     }
+    # Sanic passes its "sanic.websockets" logger to websockets' ServerProtocol,
+    # so "connection closed" (websockets >= 17, INFO) and Sanic's own
+    # "Websocket timed out waiting for pong" (WARNING) both emit via
+    # sanic.websockets, not websockets.server. Raise it to ERROR so these
+    # routine disconnect messages are dropped while real errors still show.
+    # Patch the config defaults too, so the level survives Sanic's dictConfig.
+    LOGGING_CONFIG_DEFAULTS["loggers"]["sanic.websockets"]["level"] = "ERROR"
+    logging.getLogger("sanic.websockets").setLevel(logging.ERROR)
+    # Preview worker timeouts are already annotated in the access log extra;
+    # keep the pool's own warnings quiet so they are not logged twice.
+    logging.getLogger("mediapreview.pool").setLevel(logging.ERROR)
     # Also reformat any handlers already attached (covers the initial Sanic() call)
     for name in ("sanic.root", "sanic.error", "sanic.server", "sanic.websockets"):
         for handler in logging.getLogger(name).handlers:
             handler.setFormatter(_EmojiFormatter())
+
+
+def reset_sanic_log_levels() -> None:
+    """Force Sanic's loggers back to INFO in debug/dev mode.
+
+    Debug mode enables DEBUG on sanic.root at runtime
+    (ApplicationState.set_mode calls logger.setLevel(DEBUG)), which unleashes
+    useless noise like the 'Error Page:' content-negotiation messages. Call
+    from before_server_start so the override lands after Sanic's own setup.
+    """
+    for name in ("sanic.root", "sanic.error", "sanic.server"):
+        logging.getLogger(name).setLevel(logging.INFO)

cista/sso.py

@@ -10,8 +10,10 @@ Environment variables:
 """
 
 import asyncio
+import hashlib
 import os
 import re
+from time import time
 
 import httpx
 import websockets
@@ -62,6 +64,40 @@ async def close_client():
         _client = None
 
 
+# In-memory cache for successful SSO /auth/api/validate responses.
+# Keyed by (credential hash, validation URL) so that entries for different
+# perms/renew flags coexist and all entries for a credential can be purged
+# on logout.
+_VALIDATE_CACHE_TTL = 10
+_validate_cache: dict[tuple[str, str], tuple[float, dict]] = {}
+
+
+def _validate_credential_key(request) -> str:
+    """Return a stable key for the credential material in the request."""
+    cookie = request.headers.get("cookie", "")
+    authorization = request.headers.get("authorization", "")
+    return hashlib.sha256(f"{cookie}\x00{authorization}".encode()).hexdigest()
+
+
+def _cleanup_validate_cache() -> None:
+    """Drop expired cache entries."""
+    now = time()
+    for key, (timestamp, _) in list(_validate_cache.items()):
+        if now - timestamp >= _VALIDATE_CACHE_TTL:
+            del _validate_cache[key]
+
+
+def invalidate_validation_cache(request) -> None:
+    """Remove cached SSO validations for the credentials carried by *request*.
+
+    Called after a logout request so the next request is forced to the
+    backend instead of being served from a stale success cache.
+    """
+    credential_key = _validate_credential_key(request)
+    for key in [key for key in _validate_cache if key[0] == credential_key]:
+        del _validate_cache[key]
+
+
 async def validate_sso_request(
     request, *, perm: str = "cista:login", renew: bool = True
 ) -> dict | None:
@@ -105,6 +141,17 @@ async def validate_sso_request(
     if not renew:
         url += "&renew=0"
 
+    credential_key = _validate_credential_key(request)
+    cache_key = (credential_key, url)
+
+    cached = _validate_cache.get(cache_key)
+    if cached is not None:
+        timestamp, data = cached
+        if time() - timestamp < _VALIDATE_CACHE_TTL:
+            request.ctx.sso_user = data
+            return data
+        del _validate_cache[cache_key]
+
     try:
         response = await client.post(
             url,
@@ -121,6 +168,8 @@ async def validate_sso_request(
                 request.ctx.sso_user = {}
                 return {}
             else:
+                _cleanup_validate_cache()
+                _validate_cache[cache_key] = (time(), data)
                 return data
 
         try:

cista/util/apphelpers.py

@@ -1,14 +1,15 @@
+import asyncio
 import time
 from functools import wraps
 
 import msgspec
 import websockets.exceptions
 from sanic import errorpages
-from sanic.exceptions import SanicException
+from sanic.exceptions import SanicException, Unauthorized
 from sanic.log import logger
 from sanic.response import raw, redirect
 
-from cista import auth
+from cista import auth, config, session, sharefs, sso, watching
 from cista.protocol import ErrorMsg
 from cista.sanic_logging import log_ws_close, log_ws_open
 
@@ -101,3 +102,101 @@ def websocket_wrapper(handler):
             log_ws_close(ws_id, close_code, duration, extra=close_extra)
 
     return wrapper
+
+
+class StopError(Exception):
+    """Used internally to end a watch websocket's task group cleanly."""
+
+
+async def get_watch_user_info(request):
+    """Return the current user info for a watch websocket, re-validating auth.
+
+    Handles all three auth modes:
+      - Paskia/SSO: re-validates with the auth backend (cache-friendly)
+      - Built-in: re-reads the local session cookie from the live store
+      - Public: returns None when no session is present
+
+    Raises Unauthorized/Forbidden in non-public mode when the session is gone.
+    """
+    # Long-lived API/share tokens are validated once at handshake; re-checking
+    # them on every message would add unnecessary backend calls.
+    if getattr(request.ctx, "auth_token", None) is not None:
+        return None
+
+    if sso.paskia_enabled():
+        try:
+            await sso.validate_sso_request(request, renew=False)
+        except SanicException:
+            if config.config.public:
+                return None
+            raise
+        sso_user = getattr(request.ctx, "sso_user", None) or {}
+        if sso_user:
+            ctx = sso_user.get("ctx", {})
+            perms = ctx.get("permissions", [])
+            return {
+                "username": ctx.get("user", {}).get("display_name", ""),
+                "privileged": "cista:admin" in perms,
+            }
+        return None
+
+    s = session.get(request)
+    if s:
+        user = config.config.users.get(s.get("username"))
+        if user:
+            return {"username": s["username"], "privileged": user.privileged}
+
+    if config.config.public:
+        return None
+
+    raise Unauthorized("Login required", "cookie", quiet=True)
+
+
+async def _check_watch_auth_or_stop(request, ws) -> None:
+    """Re-validate watch auth; on failure send an error and raise StopError."""
+    try:
+        await get_watch_user_info(request)
+    except SanicException as exc:
+        # Match the error format used by websocket_wrapper
+        message = f"⚠️ {str(exc) or 'Authentication error'}"
+        await asend(
+            ws,
+            ErrorMsg(
+                {"code": exc.status_code, "message": message, **(exc.context or {})}
+            ),
+        )
+        raise StopError from None
+
+
+async def run_auth_checked_watch(request, ws, queue, share_token) -> None:
+    """Run the watch websocket loop with per-message and periodic auth checks.
+
+    Messages are forwarded from *queue* to *ws*. Auth is re-checked before each
+    message (hitting the SSO cache in the common case) and every 10 seconds when
+    idle, so a session invalidated on the backend does not stay open forever.
+    """
+
+    async def consume() -> None:
+        while True:
+            item = await queue.get()
+            await _check_watch_auth_or_stop(request, ws)
+            if share_token is None or (
+                isinstance(item, str) and item.startswith('{"space"')
+            ):
+                await ws.send(item)
+            else:
+                await ws.send(
+                    watching.format_root(sharefs.build_virtual_root(share_token))
+                )
+
+    async def idle_checker() -> None:
+        while True:
+            await asyncio.sleep(10)
+            await _check_watch_auth_or_stop(request, ws)
+
+    try:
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(consume())
+            tg.create_task(idle_checker())
+    except* StopError:
+        pass

cista/util/logformat.py

@@ -0,0 +1,37 @@
+"""Shared log formatting helpers with no Sanic dependency.
+
+Used by the main process (cista.sanic_logging) and by the preview worker
+subprocess, which must not import Sanic.
+"""
+
+import logging
+import unicodedata
+
+LEVEL_EMOJI = {
+    logging.DEBUG: "🔍",
+    logging.INFO: "ℹ️",  # noqa: RUF001
+    logging.WARNING: "⚠️",
+    logging.ERROR: "🛑",
+    logging.CRITICAL: "🛑",
+}
+
+
+def display_width(text: str) -> int:
+    return sum(
+        1 + (unicodedata.east_asian_width(c) in "FW")
+        for c in text
+        if unicodedata.category(c) != "Mn"
+    )
+
+
+def format_level_prefix(levelno: int) -> str:
+    emoji = LEVEL_EMOJI.get(levelno, "▪️")
+    prefix = f"{emoji} "
+    return prefix + (" " * max(0, 3 - display_width(prefix)))
+
+
+class EmojiFormatter(logging.Formatter):
+    """Compact formatter: emoji + message, no timestamp/level text/logger name."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        return format_level_prefix(record.levelno) + record.getMessage()

docker-compose.yml

@@ -1,7 +1,7 @@
 services:
   onlyoffice:
     build:
-      context: ./docker/onlyoffice-converter-patch
+      context: ./mediapreview/mediapreview/docker
       args:
         ONLYOFFICE_VERSION: "9.3.1"
     container_name: onlyoffice

docker/onlyoffice-converter-patch/Dockerfile

@@ -1,70 +0,0 @@
-# Patched OnlyOffice Document Server with configurable converter worker count.
-#
-# The Community Edition hardcodes the document converter to 1 worker,
-# which creates a severe bottleneck under concurrent load.
-# This image patches the open-source license.js to spawn a configurable
-# number of converter workers (default 8).
-#
-# Build:
-#   docker build -t onlyoffice-cista docker/onlyoffice-converter-patch
-#
-# Run:
-#   docker run -d -p 8988:80 \
-#     -e WORKERS=16 \
-#     -e JWT_SECRET=your-strong-secret \
-#     --name onlyoffice onlyoffice-cista
-#
-# JWT:
-#   Set JWT_SECRET to the same value you pass to Cista as ONLYOFFICE_JWT_SECRET.
-#   OnlyOffice will enable token validation automatically.
-#
-# The ONLYOFFICE_VERSION build arg lets you target a specific release.
-
-ARG ONLYOFFICE_VERSION=9.3.1
-
-FROM onlyoffice/documentserver:${ONLYOFFICE_VERSION}
-
-# Prevent interactive apt prompts
-ENV DEBIAN_FRONTEND=noninteractive
-
-# Install Node.js, npm, and git so we can run the FileConverter from source.
-RUN apt-get update -qq && \
-    apt-get install -y -qq --no-install-recommends \
-        nodejs \
-        npm \
-        git \
-        ca-certificates && \
-    rm -rf /var/lib/apt/lists/*
-
-# Clone the open-source server components (shallow, ~15 MB).
-# The master branch is used because the Linux/web tags are not published
-# in the server repo; the license.js file has been stable for years.
-RUN git clone --depth 1 https://github.com/ONLYOFFICE/server.git /opt/oo-server
-
-# Patch license.js so the converter worker count is read from an env var
-# instead of being hardcoded to 1.
-RUN sed -i \
-    's/count: 1,/count: parseInt(process.env.WORKERS, 10) || 8,/' \
-    /opt/oo-server/Common/sources/license.js
-
-# Install npm dependencies for the modules the FileConverter touches.
-# DocService deps are also needed because converter.js pulls in baseConnector.
-RUN cd /opt/oo-server/Common      && npm ci --no-audit --no-fund
-RUN cd /opt/oo-server/FileConverter && npm ci --no-audit --no-fund
-RUN cd /opt/oo-server/DocService  && npm ci --no-audit --no-fund
-
-# Back up the compiled pkg binary and replace it with our wrapper.
-RUN mv /var/www/onlyoffice/documentserver/server/FileConverter/converter \
-       /var/www/onlyoffice/documentserver/server/FileConverter/converter.orig
-
-COPY converter-wrapper.sh /var/www/onlyoffice/documentserver/server/FileConverter/converter
-RUN chmod +x /var/www/onlyoffice/documentserver/server/FileConverter/converter
-
-# Default worker count (override at runtime with -e WORKERS=16).
-ENV WORKERS=8
-
-# Use our custom entrypoint to persist the env var to a file that the
-# non-root converter process (user=ds) can read.
-COPY entrypoint.sh /app/ds/run-document-server-patched.sh
-RUN chmod +x /app/ds/run-document-server-patched.sh
-ENTRYPOINT ["/app/ds/run-document-server-patched.sh"]

docker/onlyoffice-converter-patch/converter-wrapper.sh

@@ -1,19 +0,0 @@
-#!/bin/bash
-# Wrapper that runs the OnlyOffice FileConverter from patched Node.js source.
-# Replaces the compiled pkg binary shipped with the Community Edition.
-
-# The env var is not passed through supervisor to the 'ds' user, so we read
-# it from a file written by the custom entrypoint.
-if [ -z "${WORKERS}" ] && [ -r /tmp/oo-converter-workers.txt ]; then
-  export WORKERS=$(cat /tmp/oo-converter-workers.txt)
-fi
-
-cd /opt/oo-server/FileConverter || exit 1
-
-export NODE_ENV=production-linux
-export NODE_CONFIG_DIR=/etc/onlyoffice/documentserver
-export NODE_DISABLE_COLORS=1
-export APPLICATION_NAME=onlyoffice
-export LD_LIBRARY_PATH=/var/www/onlyoffice/documentserver/server/FileConverter/bin
-
-exec node sources/convertermaster.js "$@"

docker/onlyoffice-converter-patch/entrypoint.sh

@@ -1,8 +0,0 @@
-#!/bin/bash
-# Custom entrypoint that persists WORKERS to a file readable by
-# the non-root user that supervisor uses to run the converter.
-
-echo "${WORKERS:-8}" > /tmp/oo-converter-workers.txt
-chmod 644 /tmp/oo-converter-workers.txt
-
-exec /app/ds/run-document-server.sh "$@"

frontend/src/components/HeaderMain.vue

@@ -28,9 +28,9 @@
     </template>
     <div v-if="!props.editorMode && showSortHints" class="sort-hints">
       <span class="sort-label">Order</span>
-      <span class="keycap">1</span>
-      <span class="keycap">2</span>
-      <span class="keycap">3</span>
+      <button type="button" class="keycap" aria-label="Alphabetical order" @click="store.sort('name')">1</button>
+      <button type="button" class="keycap" aria-label="Newest first" @click="store.sort('modified')">2</button>
+      <button type="button" class="keycap" aria-label="Largest first" @click="store.sort('size')">3</button>
     </div>
     <SvgButton
       v-if="props.editorMode"
@@ -314,6 +314,14 @@ onUnmounted(() => {
   border-radius: 0.3em;
   padding: 0 0.45em;
   line-height: 1.4;
+  cursor: pointer;
+  transition: all 0.2s ease;
+}
+.keycap:hover,
+.keycap:focus {
+  background: #e6e6e6;
+  border-color: #aaa;
+  transform: scale(1.05);
 }
 @media screen and (min-width: 800px) {
   .sort-hints {

frontend/src/stores/main.ts

@@ -79,6 +79,7 @@ export const useMainStore = defineStore('main', {
     connected: false,
     authInProgress: false,
     cursor: '' as string,
+    lastSearchLoc: '' as string,
     server: {} as Record<string, any> & {
       public?: boolean
       paskia?: boolean
@@ -159,6 +160,10 @@ export const useMainStore = defineStore('main', {
       this.docVersion++
       // Sync documents to search worker
       this.syncSearchWorker()
+      // Re-run the current search against the updated file list
+      if (this.query) {
+        this.search(this.query, this.lastSearchLoc)
+      }
     },
     /** Patch aspect ratios on existing docs from a server ar update message */
     updateAr(arMap: Record<string, number>) {
@@ -268,6 +273,7 @@ export const useMainStore = defineStore('main', {
 
       // Update query immediately so watchers know we're handling this
       this.query = query
+      this.lastSearchLoc = loc
 
       // Cancel pending timers
       if (loadingTimer) {

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "html5tagger>=1.3.0",
     "httpx>=0.28.0",
     "inotify>=0.2.12",
+    "mediapreview[standard]>=0.2.0",
     "msgspec>=0.19.0",
     "natsort>=8.4.0",
     "numpy>=2.3.2",
@@ -77,7 +78,7 @@ docs = [
 source = "vcs"
 
 [tool.hatch.build]
-artifacts = ["cista/frontend-build", "cista/docker"]
+artifacts = ["cista/frontend-build"]
 targets.sdist.hooks.custom.path = "scripts/fastapi-vue/buildhook.py"
 targets.sdist.include = [
     "/cista",
@@ -161,7 +162,7 @@ ignore = [
     "TRY003", # exception-message strictness too noisy on legacy handlers
 ]
 isort.known-first-party = ["cista"]
-per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001"]
+per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001", "SLF001"]
 per-file-ignores."scripts/*" = ["T20"]
 
 [dependency-groups]

tests/test_sso_cache.py

@@ -0,0 +1,142 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from sanic.exceptions import Unauthorized
+
+from cista import sso
+
+
+def _make_request(cookie: str = "", authorization: str = ""):
+    req = SimpleNamespace()
+    req.headers = {}
+    if cookie:
+        req.headers["cookie"] = cookie
+    if authorization:
+        req.headers["authorization"] = authorization
+    req.client_ip = "127.0.0.1"
+    req.host = "test.local"
+    req.scheme = "http"
+    req.ctx = SimpleNamespace()
+    return req
+
+
+@pytest.fixture(autouse=True)
+def _reset_sso_cache_and_client(monkeypatch):
+    """Clear the SSO validation cache and shared client between tests."""
+    sso._validate_cache.clear()
+    sso._client = None
+    monkeypatch.setenv("PASKIA_BACKEND_URL", "http://test-paskia.local")
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+    yield
+    sso._validate_cache.clear()
+
+
+@pytest.fixture
+def mock_client(monkeypatch):
+    client = AsyncMock()
+    client.is_closed = False
+    client.headers = {}
+    monkeypatch.setattr(sso, "_client", client)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_caches_successful_responses(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    data1 = await sso.validate_sso_request(req)
+    data2 = await sso.validate_sso_request(req)
+
+    assert data1 == {"user": "alice"}
+    assert data2 == data1
+    assert mock_client.post.call_count == 1
+    assert req.ctx.sso_user == {"user": "alice"}
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_does_not_cache_errors(mock_client):
+    req = _make_request(cookie="session=bad")
+    mock_client.post.return_value = httpx.Response(401, json={"detail": "nope"})
+
+    with pytest.raises(Unauthorized):
+        await sso.validate_sso_request(req)
+    with pytest.raises(Unauthorized):
+        await sso.validate_sso_request(req)
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_cache_is_per_credential(mock_client):
+    req_alice = _make_request(cookie="session=alice")
+    req_bob = _make_request(cookie="session=bob")
+    responses = {
+        "alice": httpx.Response(200, json={"user": "alice"}),
+        "bob": httpx.Response(200, json={"user": "bob"}),
+    }
+
+    def side_effect(*args, **kwargs):
+        cookie = kwargs.get("headers", {}).get("cookie", "")
+        if "alice" in cookie:
+            return responses["alice"]
+        return responses["bob"]
+
+    mock_client.post.side_effect = side_effect
+
+    assert await sso.validate_sso_request(req_alice) == {"user": "alice"}
+    assert await sso.validate_sso_request(req_bob) == {"user": "bob"}
+    assert await sso.validate_sso_request(req_alice) == {"user": "alice"}
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_cache_is_per_permission(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    await sso.validate_sso_request(req, perm="cista:login")
+    await sso.validate_sso_request(req, perm="cista:admin")
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_invalidate_validation_cache_forces_backend_call(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    await sso.validate_sso_request(req)
+    sso.invalidate_validation_cache(req)
+    await sso.validate_sso_request(req)
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_invalidate_validation_cache_only_affects_same_credentials(mock_client):
+    alice = _make_request(cookie="session=alice")
+    bob = _make_request(cookie="session=bob")
+    responses = {
+        "alice": httpx.Response(200, json={"user": "alice"}),
+        "bob": httpx.Response(200, json={"user": "bob"}),
+    }
+
+    def side_effect(*args, **kwargs):
+        cookie = kwargs.get("headers", {}).get("cookie", "")
+        return responses["alice"] if "alice" in cookie else responses["bob"]
+
+    mock_client.post.side_effect = side_effect
+
+    await sso.validate_sso_request(alice)
+    await sso.validate_sso_request(bob)
+    sso.invalidate_validation_cache(alice)
+
+    assert await sso.validate_sso_request(alice) == {"user": "alice"}
+    assert await sso.validate_sso_request(bob) == {"user": "bob"}
+
+    # Alice is re-fetched; bob is still cached.
+    assert mock_client.post.call_count == 3

tests/test_watch_auth.py

@@ -0,0 +1,211 @@
+import asyncio
+import contextlib
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from sanic.exceptions import Unauthorized
+
+from cista import auth, config, session, sso
+from cista.util.apphelpers import (
+    get_watch_user_info,
+    run_auth_checked_watch,
+)
+
+
+def _make_request(cookie: str = "", auth_token=None):
+    req = SimpleNamespace()
+    req.headers = {}
+    req.cookies = {}
+    if cookie:
+        req.headers["cookie"] = cookie
+        for part in cookie.split(";"):
+            k, _, v = part.strip().partition("=")
+            req.cookies[k] = v
+    req.ctx = SimpleNamespace()
+    if auth_token:
+        req.ctx.auth_token = auth_token
+    return req
+
+
+@pytest.fixture(autouse=True)
+def _reset(tmp_path, monkeypatch):
+    alice = config.User()
+    auth.set_password(alice, "secret")
+    admin = config.User(privileged=True)
+    auth.set_password(admin, "admin-secret")
+    config.config = config.Config(
+        path=tmp_path,
+        listen=":0",
+        public=False,
+        users={"alice": alice, "admin": admin},
+    )
+    session._sessions.clear()
+    sso._validate_cache.clear()
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "")
+    yield
+    session._sessions.clear()
+    sso._validate_cache.clear()
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_valid_session():
+    token = "valid-token"
+    session.put(token, "alice")
+    req = _make_request(cookie=f"cista={token}")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "alice", "privileged": False}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_admin():
+    token = "admin-token"
+    session.put(token, "admin")
+    req = _make_request(cookie=f"cista={token}")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "admin", "privileged": True}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_invalid_session_raises():
+    req = _make_request(cookie="cista=bad-token")
+
+    with pytest.raises(Unauthorized):
+        await get_watch_user_info(req)
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_public_no_session():
+    config.config.public = True
+    req = _make_request()
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_valid(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+
+    async def mock_validate(request, *, renew=True):
+        request.ctx.sso_user = {
+            "ctx": {
+                "user": {"display_name": "alice"},
+                "permissions": ["cista:login", "cista:admin"],
+            }
+        }
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "alice", "privileged": True}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_nonpublic_invalid_raises(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+
+    async def mock_validate(request, *, renew=True):
+        raise Unauthorized("Session expired", quiet=True)
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    with pytest.raises(Unauthorized):
+        await get_watch_user_info(req)
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_public_invalid_returns_none(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+    config.config.public = True
+
+    async def mock_validate(request, *, renew=True):
+        raise Unauthorized("Session expired", quiet=True)
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_run_auth_checked_watch_forwards_messages_while_valid():
+    token = "valid-token"
+    session.put(token, "alice")
+    req = _make_request(cookie=f"cista={token}")
+    ws = AsyncMock()
+    q = asyncio.Queue()
+
+    async def producer():
+        await q.put('{"space":{}}')
+        await q.put('{"update":[]}')
+        # Keep consumer alive briefly, then invalidate.
+        await asyncio.sleep(0.05)
+        session._sessions.pop(token, None)
+        await q.put('{"update":[]}')
+
+    await asyncio.wait_for(
+        asyncio.gather(producer(), run_auth_checked_watch(req, ws, q, None)),
+        timeout=1.0,
+    )
+
+    calls = [c.args[0] for c in ws.send.call_args_list]
+    assert calls[0] == '{"space":{}}'
+    assert calls[1] == '{"update":[]}'
+    assert '"error"' in calls[2]
+    assert len(calls) == 3
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_token_auth_skips_revalidation():
+    """Token-based auth is considered valid without re-checking the token."""
+    token_id = "api-token"
+    config.config.tokens[token_id] = config.Token(
+        key=token_id, username="alice", kind="api", mode="rw"
+    )
+    req = _make_request(auth_token=config.config.tokens[token_id])
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_run_auth_checked_watch_token_auth_does_not_send_errors():
+    """Token-based sockets keep forwarding messages without re-validating."""
+    token_id = "api-token"
+    config.config.tokens[token_id] = config.Token(
+        key=token_id, username="alice", kind="api", mode="rw"
+    )
+    req = _make_request(auth_token=config.config.tokens[token_id])
+    ws = AsyncMock()
+    q = asyncio.Queue()
+
+    async def producer():
+        await q.put('{"space":{}}')
+        await q.put('{"update":[]}')
+        # Deleting the token should not affect the already-open websocket.
+        await asyncio.sleep(0.05)
+        del config.config.tokens[token_id]
+
+    runner = asyncio.create_task(run_auth_checked_watch(req, ws, q, None))
+    await asyncio.wait_for(producer(), timeout=1.0)
+    await asyncio.sleep(0.05)
+    runner.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await runner
+
+    calls = [c.args[0] for c in ws.send.call_args_list]
+    assert calls[0] == '{"space":{}}'
+    assert calls[1] == '{"update":[]}'
+    assert not any('"error"' in c for c in calls)