Separate OnlyOffice async handling, add office_previews WS flag, framed worker protocol

- Move OnlyOffice conversion out of worker subprocesses into async event loop
  using httpx.AsyncClient with shared client and clean shutdown hook
- Add OOConversionManager with in-flight deduplication (asyncio.Future) and
  configurable concurrency limit (OO_MAX_CONCURRENT=2)
- Add 10s total timeout for office previews via asyncio.wait_for
- Return 503 when OnlyOffice is unavailable, 504 on timeout
- Remove office handling from worker dispatch(); workers now only do
  images, video, and PDFs
- Replace PNG tempfile bridge with framed binary input protocol on worker
  stdin: (json_size)(data_size)(json)(raw_data)
- Add process_image_buffer() for in-memory AVIF conversion via pyvips
- Add office_previews to WS server message, cached every 30s from OO
  availability check
- Frontend gates only office document previews on office_previews flag;
  images, video, PDFs remain unconditional
- Change default OnlyOffice port from 8080 to 8988

cista/api.py

@@ -55,6 +55,8 @@ async def watch(req, ws):
             "privileged": req.ctx.user.privileged,
         }
 
+    from cista import onlyoffice
+
     await ws.send(
         msgspec.json.encode(
             {
@@ -63,6 +65,7 @@ async def watch(req, ws):
                     "version": __version__,
                     "public": config.config.public,
                     "paskia": sso.paskia_enabled(),
+                    "office_previews": await onlyoffice.is_available_cached(),
                 },
                 "user": user_info,
             }

cista/app.py

@@ -16,7 +16,7 @@ from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
 from zstandard import ZstdCompressor
 
-from cista import auth, config, fileserver, preview, session, sharefs, sso, watching
+from cista import auth, config, fileserver, onlyoffice, preview, session, sharefs, sso, watching
 from cista.api import bp
 from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.sanic_logging import (
@@ -130,6 +130,7 @@ async def main_start(app):
 @app.before_server_stop
 async def main_stop(app):
     watching.stop(app)
+    await onlyoffice.close_oo_client()
     await shutdown_preview_workers()
     app.ctx.threadexec.shutdown()
     app.ctx.zipexec.shutdown(cancel_futures=True)

cista/onlyoffice.py

@@ -10,6 +10,7 @@ Environment requirements:
       reachable from the container (usually the docker bridge IP).
 """
 
+import asyncio
 import json
 import os
 import socket
@@ -23,6 +24,7 @@ from pathlib import Path
 from time import perf_counter
 from urllib.parse import quote
 
+import httpx
 import jwt
 from sanic.log import logger
 
@@ -31,8 +33,11 @@ from sanic.log import logger
 # ---------------------------------------------------------------------------
 
 
+_httpx_client: httpx.AsyncClient | None = None
+
+
 def _get_onlyoffice_url() -> str:
-    return os.environ.get("ONLYOFFICE_URL", "http://localhost:8080")
+    return os.environ.get("ONLYOFFICE_URL", "http://localhost:8988")
 
 
 def _get_jwt_secret() -> str | None:
@@ -62,6 +67,27 @@ def _get_callback_host() -> str:
     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
 # ---------------------------------------------------------------------------
@@ -69,14 +95,42 @@ def _get_callback_host() -> str:
 
 def is_available() -> bool:
     """Return True if the configured OnlyOffice Document Server is reachable."""
-    url = _get_onlyoffice_url()
+    url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx"
     try:
         with urllib.request.urlopen(url, timeout=3) as resp:  # noqa: S310
-            return resp.status == 200
+            return resp.status in (200, 405)  # 405 Method Not Allowed is fine, means endpoint exists
     except Exception:
         return False
 
 
+async def is_available_async(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=timeout)
+        return response.status_code in (200, 405)
+    except Exception:
+        return False
+
+
+_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
 # ---------------------------------------------------------------------------
@@ -121,7 +175,7 @@ def _build_jwt_token(payload: dict) -> str | None:
     return jwt.encode(payload, secret, algorithm="HS256")
 
 
-def convert_to_png(file_path: Path, timeout: float = 30.0) -> bytes:
+def convert_to_png(file_path: Path, timeout: float = 5.0) -> bytes:
     """Convert *file_path* to PNG using OnlyOffice Document Server.
 
     Returns the PNG bytes. Raises RuntimeError on failure.
@@ -182,3 +236,67 @@ def convert_to_png(file_path: Path, timeout: float = 30.0) -> bytes:
             return png_resp.read()
     finally:
         httpd.shutdown()
+
+
+async def convert_to_png_async(file_path: Path, 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_{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=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=timeout)
+        png_response.raise_for_status()
+        return png_response.content
+    finally:
+        await asyncio.to_thread(httpd.shutdown)

cista/preview.py

@@ -10,7 +10,7 @@ import urllib.parse
 from collections import OrderedDict
 from dataclasses import dataclass
 from multiprocessing import cpu_count
-from pathlib import PurePosixPath
+from pathlib import Path, PurePosixPath
 from time import perf_counter
 from urllib.parse import unquote
 from wsgiref.handlers import format_date_time
@@ -112,24 +112,25 @@ class _PreviewWorker:
     def __init__(self, proc: asyncio.subprocess.Process):
         self.proc = proc
 
-    async def request(self, filepath, quality: int, maxsize: int, maxzoom: float):
+    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")
 
-        line = (
-            msgspec.json.encode(
-                PreviewRequest(
-                    path=str(filepath),
-                    quality=quality,
-                    maxsize=maxsize,
-                    maxzoom=maxzoom,
-                )
+        meta = msgspec.json.encode(
+            PreviewRequest(
+                path=str(filepath),
+                quality=quality,
+                maxsize=maxsize,
+                maxzoom=maxzoom,
             )
-            + b"\n"
         )
-        self.proc.stdin.write(line)
+        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)
@@ -270,7 +271,9 @@ class _PreviewWorkerPool:
         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):
+    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()
@@ -281,7 +284,7 @@ class _PreviewWorkerPool:
                 _preview_job_priority(filepath),
                 self._seq,
                 future,
-                (filepath, quality, maxsize, maxzoom),
+                (filepath, quality, maxsize, maxzoom, data),
             )
         )
         return await future
@@ -355,6 +358,10 @@ class PreviewTimeoutError(Exception):
     """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
 
 
+class OnlyOfficeUnavailableError(Exception):
+    """Raised when the OnlyOffice Document Server is not reachable."""
+
+
 class PreviewError(Exception):
     """Raised when the preview subprocess exits with a non-zero status."""
 
@@ -370,14 +377,98 @@ class PreviewError(Exception):
         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 = 2
+
+
+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._lock = asyncio.Lock()
+
+    async def convert(self, filepath: Path) -> bytes:
+        """Return PNG bytes for *filepath*, deduplicating concurrent requests."""
+        stat = 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
+                asyncio.create_task(self._do_convert(filepath, key, future))
+
+        return await future
+
+    async def _do_convert(
+        self, filepath: Path, key: str, future: asyncio.Future[bytes]
+    ) -> None:
+        try:
+            async with self._semaphore:
+                oo = _get_onlyoffice()
+                if oo is False:
+                    raise OnlyOfficeUnavailableError("OnlyOffice is not installed")
+                png_bytes = await oo.convert_to_png_async(filepath, timeout=5.0)
+        except Exception as e:
+            future.set_exception(e)
+            async with self._lock:
+                self._in_flight.pop(key, None)
+        else:
+            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."""
+    oo = _get_onlyoffice()
+    if oo is False:
+        raise OnlyOfficeUnavailableError("OnlyOffice is not installed")
+    if not await oo.is_available_async():
+        raise OnlyOfficeUnavailableError("OnlyOffice Document Server is not reachable")
+
+    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
+    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)
+    return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data)
 
 
 DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
@@ -492,9 +583,19 @@ async def preview(req, path):
 
     # Generate preview
     try:
-        img, preview_resp = await _run_preview_process(
-            filepath, quality, maxsize, maxzoom
-        )
+        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,
+            )
+        else:
+            img, preview_resp = await _run_preview_process(
+                filepath, quality, maxsize, maxzoom
+            )
+    except asyncio.TimeoutError:
+        return empty(504)
+    except OnlyOfficeUnavailableError:
+        return empty(503)
     except PreviewTimeoutError:
         return empty(504)
     except PreviewError as e:
@@ -539,18 +640,16 @@ async def preview(req, path):
     return raw(img, headers=headers)
 
 
-def dispatch(path, quality, maxsize, maxzoom):
+def dispatch(path, quality, maxsize, maxzoom, data=None):
     backend = "unknown"
     try:
+        if data is not None:
+            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)
-        if suffix in OFFICE_PREVIEW_SUFFIXES:
-            backend = "onlyoffice"
-            return process_office(
-                path, quality=quality, maxsize=maxsize, maxzoom=maxzoom
-            )
         mime_type, _ = mimetypes.guess_type(path.name)
         if mime_type and mime_type.startswith("video/"):
             backend = "video"
@@ -592,6 +691,29 @@ def process_image_pyvips(path, *, maxsize, quality):
     )
 
 
+def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
+    t_start = perf_counter()
+    img = pyvips.Image.new_from_buffer(data, "")
+    img = img.autorot()
+    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)],
+    )
+
+
 def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
     t_load_start = perf_counter()
     pdf = fitz.open(path)

cista/preview_worker.py

@@ -2,9 +2,12 @@
 
 Two modes are supported:
 1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom.
-2) Long-lived mode: read JSONL commands from stdin and write framed responses.
+2) Long-lived mode: read framed requests from stdin and write framed responses.
 
-Framed response format:
+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).
 """
@@ -40,6 +43,30 @@ _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
@@ -70,18 +97,18 @@ def _run_loop() -> None:
     from cista.preview import dispatch
 
     while True:
-        line = sys.stdin.buffer.readline()
-        if not line:
+        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):
-                req = _dec_req.decode(line)
                 result, resp = dispatch(
-                    Path(req.path), req.quality, req.maxsize, req.maxzoom
+                    Path(req.path), req.quality, req.maxsize, req.maxzoom, data
                 )
             if not resp.ok:
                 captured = stderr_capture.getvalue().strip()