Uploads by PUT range requests rather than WS, remove dead code WS handlers (#9)

Increases block size to 16 MiB and adjusts progress display to work smoothly with that. Also removes download WS that was already unused. Provides faster upload speed than over WS.

cista/api.py

@@ -1,5 +1,4 @@
 import asyncio
-import typing
 from pathlib import PurePosixPath
 from secrets import token_bytes
 
@@ -9,7 +8,7 @@ from sanic.exceptions import BadRequest
 
 from cista import __version__, auth, config, sso, watching
 from cista.fileio import FileServer
-from cista.protocol import ControlTypes, FileRange, StatusMsg
+from cista.protocol import ControlTypes, StatusMsg
 from cista.util.apphelpers import asend, websocket_wrapper
 
 bp = Blueprint("api", url_prefix="/api")
@@ -26,65 +25,6 @@ async def stop_fileserver(app):
     await fileserver.stop()
 
 
-@bp.websocket("upload")
-@websocket_wrapper
-async def upload(req, ws):
-    alink = fileserver.alink
-    while True:
-        req = None
-        text = await ws.recv()
-        if not isinstance(text, str):
-            raise ValueError(
-                f"Expected JSON control, got binary len(data) = {len(text)}",
-            )
-        req = msgspec.json.decode(text, type=FileRange)
-        pos = req.start
-        while True:
-            data = await ws.recv()
-            if not isinstance(data, bytes):
-                break
-            if len(data) > req.end - pos:
-                raise ValueError(
-                    f"Expected up to {req.end - pos} bytes, got {len(data)} bytes"
-                )
-            sentsize = await alink(("upload", req.name, pos, data, req.size))
-            pos += typing.cast(int, sentsize)
-            if pos >= req.end:
-                break
-        if pos != req.end:
-            d = f"{len(data)} bytes" if isinstance(data, bytes) else data
-            raise ValueError(f"Expected {req.end - pos} more bytes, got {d}")
-        # Signal the watcher about the uploaded file and its parent directories
-        path = PurePosixPath(req.name)
-        watching.notify_change(path, *path.parents)
-        # Report success
-        res = StatusMsg(status="ack", req=req)
-        await asend(ws, res)
-
-
-@bp.websocket("download")
-@websocket_wrapper
-async def download(req, ws):
-    alink = fileserver.alink
-    while True:
-        req = None
-        text = await ws.recv()
-        if not isinstance(text, str):
-            raise ValueError(
-                f"Expected JSON control, got binary len(data) = {len(text)}",
-            )
-        req = msgspec.json.decode(text, type=FileRange)
-        pos = req.start
-        while pos < req.end:
-            end = min(req.end, pos + (1 << 20))
-            data = typing.cast(bytes, await alink(("download", req.name, pos, end)))
-            await asend(ws, data)
-            pos += len(data)
-        # Report success
-        res = StatusMsg(status="ack", req=req)
-        await asend(ws, res)
-
-
 @bp.websocket("control")
 @websocket_wrapper
 async def control(req, ws):

cista/app.py

@@ -1,6 +1,8 @@
 import asyncio
 import datetime
 import mimetypes
+import re
+import time
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import cpu_count
 from pathlib import Path, PurePath, PurePosixPath
@@ -10,21 +12,27 @@ from wsgiref.handlers import format_date_time
 
 import sanic.helpers
 from blake3 import blake3
-from sanic import Blueprint, Sanic, empty, raw, redirect
+from sanic import Blueprint, Sanic, empty, json, raw, redirect
-from sanic.exceptions import Forbidden, NotFound
+from sanic.exceptions import BadRequest, Forbidden, NotFound
 from sanic.log import logger
 from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
 from zstandard import ZstdCompressor
 
 from cista import auth, config, preview, session, sso, watching
-from cista.api import bp
+from cista.preview import shutdown_preview_workers, start_preview_workers
+from cista.api import bp, fileserver
+from cista.sanic_logging import configure_access_logging, configure_main_logging, format_access_log
+from cista.sanic_logging import logger as access_logger
 from cista.util.apphelpers import handle_sanic_exception
 
 # Workaround until Sanic PR #2824 is merged
 sanic.helpers._ENTITY_HEADERS = frozenset()
 
+configure_access_logging()
+
 app = Sanic("cista", strict_slashes=True)
+configure_main_logging()
 # Register either SSO proxy or built-in auth routes based on PASKIA_BACKEND_URL
 if sso.paskia_enabled():
     app.blueprint(sso.bp)  # SSO proxy for /auth/* routes
@@ -42,13 +50,12 @@ setproctitle("cista-main")
 async def main_start(app):
     config.load_config()
     setproctitle(f"cista {config.config.path.name}")
-    # Small pool for memory-intensive preview generation
-    preview_workers = max(2, min(8, cpu_count()))
     app.ctx.threadexec = ThreadPoolExecutor(
-        max_workers=preview_workers, thread_name_prefix="cista-preview"
+        max_workers=4, thread_name_prefix="cista-worker"
     )
     # Larger pool for long-running but low-memory zip operations
     app.ctx.zipexec = ThreadPoolExecutor(max_workers=32, thread_name_prefix="cista-zip")
+    await start_preview_workers()
     watching.start(app)
 
 
@@ -56,6 +63,7 @@ async def main_start(app):
 @app.before_server_stop
 async def main_stop(app):
     watching.stop(app)
+    await shutdown_preview_workers()
     app.ctx.threadexec.shutdown()
     app.ctx.zipexec.shutdown(cancel_futures=True)
     await sso.close_client()
@@ -64,6 +72,7 @@ async def main_stop(app):
 
 @app.on_request
 async def use_session(req):
+    req.ctx._log_start = time.perf_counter()
     req.ctx.session = session.get(req)
     try:
         req.ctx.username = req.ctx.session["username"]  # type: ignore
@@ -81,6 +90,27 @@ async def use_session(req):
         raise Forbidden("Invalid origin: Cross-Site requests not permitted")
 
 
+@app.on_response
+async def log_access(req, res):
+    """Log HTTP access in a clean single-line format."""
+    if req.headers.get("upgrade", "").lower() == "websocket":
+        return res
+    start = getattr(req.ctx, "_log_start", None)
+    duration_ms = (time.perf_counter() - start) * 1000 if start is not None else 0.0
+    client = req.client_ip or "-"
+    host = req.host or "-"
+    path = req.path
+    if req.query_string:
+        qs = req.query_string
+        if isinstance(qs, bytes):
+            qs = qs.decode(errors="replace")
+        path = f"{path}?{qs}"
+    extra = getattr(req.ctx, "_log_extra", None)
+    line = format_access_log(client, res.status, req.method, host, path, duration_ms, extra=extra)
+    access_logger.info(line)
+    return res
+
+
 @app.on_response
 async def forward_sso_cookies(req, res):
     """Forward Set-Cookie headers from SSO validation to client."""
@@ -98,6 +128,68 @@ def http_fileserver(app):
         """Verify access to file server routes."""
         await auth.verify(request)
 
+    @bp.put("/files/<name:path>")
+    async def upload_file_chunk(request, *args, **kwargs):
+        body = request.body
+        header = request.headers.get("content-range")
+        if header:
+            start, end, total = _parse_content_range(header, len(body))
+        else:
+            start = 0
+            end = len(body)
+            total = end
+        raw_name = kwargs.get("name")
+        if raw_name is None and args:
+            raw_name = args[0]
+        if not isinstance(raw_name, str) or not raw_name:
+            prefix = "/files/"
+            if not request.path.startswith(prefix):
+                raise BadRequest("Invalid upload path")
+            raw_name = request.path[len(prefix) :]
+        rel_name = unquote(raw_name)
+        upload_info = await asyncio.to_thread(
+            fileserver.upload_info,
+            rel_name,
+            start,
+            body,
+            total,
+        )
+        extras = []
+        chunk_len = end - start
+        whole_file = start == 0 and end == total
+        if not whole_file:
+            start_mib = _to_mib_int(start)
+            chunk_mib = _to_mib_int(chunk_len)
+            # Keep range logs compact for fixed-size upload blocks.
+            if chunk_mib == 16:
+                extras.append(f"{start_mib}MiB")
+            else:
+                extras.append(f"{start_mib}+{chunk_mib}MiB")
+        if upload_info.get("created"):
+            extras.append(f"created {_to_mib_int(total)}MiB")
+        size_before = upload_info.get("size_before")
+        size_after = upload_info.get("size_after")
+        if (
+            size_before is not None
+            and size_after is not None
+            and size_before != size_after
+        ):
+            extras.append("resized")
+        request.ctx._log_extra = " ".join(extras) if extras else None
+        path = PurePosixPath(rel_name)
+        watching.notify_change(path, *path.parents)
+        return json(
+            {
+                "status": "ack",
+                "req": {
+                    "name": rel_name,
+                    "size": total,
+                    "start": start,
+                    "end": end,
+                },
+            }
+        )
+
     bp.static(
         "/files/",
         config.config.path,
@@ -109,6 +201,30 @@ def http_fileserver(app):
 
 
 www = {}
+_CONTENT_RANGE_RE = re.compile(r"^bytes (\d+)-(\d+)/(\d+)$")
+
+
+def _parse_content_range(header: str, body_len: int) -> tuple[int, int, int]:
+    m = _CONTENT_RANGE_RE.fullmatch(header.strip())
+    if m is None:
+        raise BadRequest("Invalid Content-Range format")
+    start, end_inclusive, total = (int(v) for v in m.groups())
+    if total <= 0:
+        raise BadRequest("Invalid Content-Range total size")
+    if start > end_inclusive:
+        raise BadRequest("Invalid Content-Range range")
+    if end_inclusive >= total:
+        raise BadRequest("Content-Range exceeds total size")
+    expected_len = end_inclusive - start + 1
+    if expected_len != body_len:
+        raise BadRequest(
+            f"Content length mismatch for range: expected {expected_len}, got {body_len}"
+        )
+    return start, end_inclusive + 1, total
+
+
+def _to_mib_int(value_bytes: int) -> int:
+    return round(value_bytes / (1 << 20))
 
 
 def _load_wwwroot(www):

cista/fileio.py

@@ -1,9 +1,8 @@
-import asyncio
 import os
+import threading
 
 from cista import config
 from cista.util import filename
-from cista.util.asynclink import AsyncLink
 from cista.util.lrucache import LRUCache
 
 
@@ -62,38 +61,32 @@ class File:
 
 class FileServer:
     async def start(self):
-        self.alink = AsyncLink()
-        self.worker = asyncio.get_event_loop().run_in_executor(
-            None,
-            self.worker_thread,
-            self.alink.to_sync,
-        )
         self.cache = LRUCache(File, capacity=10, maxage=5.0)
+        self.cache_lock = threading.Lock()
+        self.file_locks: dict[str, threading.Lock] = {}
 
     async def stop(self):
-        await self.alink.stop()
-        await self.worker
-
-    def worker_thread(self, slink):
-        try:
-            for req in slink:
-                with req as (command, *args):
-                    if command == "upload":
-                        req.set_result(self.upload(*args))
-                    elif command == "download":
-                        req.set_result(self.download(*args))
-                    else:
-                        raise NotImplementedError(f"Unhandled {command=} {args}")
-        finally:
         self.cache.close()
 
-    def upload(self, name, pos, data, file_size):
+    @staticmethod
-        name = filename.sanitize(name)
+    def _stat_size(path):
-        f = self.cache[name]
+        try:
-        f.write(pos, data, file_size=file_size)
+            return os.stat(path).st_size
-        return len(data)
+        except FileNotFoundError:
+            return None
 
-    def download(self, name, start, end):
+    def upload_info(self, name, pos, data, file_size):
         name = filename.sanitize(name)
+        with self.cache_lock:
             f = self.cache[name]
-        return f[start:end]
+            lock = self.file_locks.setdefault(name, threading.Lock())
+        with lock:
+            size_before = self._stat_size(f.path)
+            f.write(pos, data, file_size=file_size)
+            size_after = self._stat_size(f.path)
+        return {
+            "written": len(data),
+            "created": size_before is None,
+            "size_before": size_before,
+            "size_after": size_after,
+        }

cista/preview.py

@@ -2,29 +2,33 @@ import asyncio
 import gc
 import io
 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 PurePosixPath
 from time import perf_counter
 from urllib.parse import unquote
 from wsgiref.handlers import format_date_time
 
+import msgspec
+
 import av
 import fitz  # PyMuPDF
 import numpy as np
-import pillow_heif
+import pyvips
-from PIL import Image
+from blake3 import blake3
 from sanic import Blueprint, empty, raw, redirect
 from sanic.exceptions import NotFound
 from sanic.log import logger
 
 from cista import auth, config
+from cista.preview_worker import PreviewRequest, PreviewResponse
 from cista.util.filename import sanitize
 
-pillow_heif.register_heif_opener()
-
 bp = Blueprint("preview", url_prefix="/preview")
 
 
@@ -70,6 +74,213 @@ class PreviewCache:
 # Global preview cache instance
 _preview_cache = PreviewCache(capacity=500)
 
+PREVIEW_TIMEOUT = 3.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):
+        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,
+                )
+            )
+            + b"\n"
+        )
+        self.proc.stdin.write(line)
+        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:
+            try:
+                self.proc.kill()
+            except ProcessLookupError:
+                pass
+            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._workers: set[_PreviewWorker] = set()
+        self._closed = False
+
+    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.DEVNULL,
+            start_new_session=True,
+        )
+        _active_procs.add(proc)
+        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 start(self) -> None:
+        for _ in range(self.size):
+            await self._add_worker()
+
+    async def run(self, filepath, quality: int, maxsize: int, maxzoom: float):
+        if self._closed:
+            raise PreviewError("preview worker pool closed")
+        worker = await self._idle.get()
+        replace = False
+        try:
+            out, resp = await asyncio.wait_for(
+                worker.request(filepath, quality, maxsize, maxzoom),
+                timeout=PREVIEW_TIMEOUT,
+            )
+            return out, resp
+        except asyncio.TimeoutError:
+            replace = True
+            logger.warning(
+                "Preview timeout (%ds) for %s", int(PREVIEW_TIMEOUT), filepath.name
+            )
+            raise PreviewTimeout(filepath.name)
+        except WorkerChecksumError:
+            replace = True
+            logger.error("Preview checksum mismatch for %s", filepath.name)
+            raise PreviewError(f"worker checksum mismatch for {filepath.name}")
+        except PreviewError:
+            raise
+        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
+            )
+            raise PreviewError(
+                f"worker protocol failure for {filepath.name}: {e}"
+            )
+        finally:
+            if replace:
+                await self._replace_worker(worker)
+            else:
+                if worker.proc.returncode is None:
+                    await self._idle.put(worker)
+                else:
+                    await self._replace_worker(worker)
+
+    async def close(self) -> None:
+        self._closed = True
+        workers = list(self._workers)
+        self._workers.clear()
+        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):
+        try:
+            proc.kill()
+        except ProcessLookupError:
+            pass
+    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):
@@ -77,16 +288,46 @@ async def verify_preview(request):
     await auth.verify(request)
 
 
-# Map EXIF Orientation value to a corresponding PIL transpose
+class PreviewTimeout(Exception):
-EXIF_ORI = {
+    """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
-    2: Image.Transpose.FLIP_LEFT_RIGHT,
+
-    3: Image.Transpose.ROTATE_180,
+
-    4: Image.Transpose.FLIP_TOP_BOTTOM,
+class PreviewError(Exception):
-    5: Image.Transpose.TRANSPOSE,
+    """Raised when the preview subprocess exits with a non-zero status."""
-    6: Image.Transpose.ROTATE_270,
+
-    7: Image.Transpose.TRANSVERSE,
+    def __init__(
-    8: Image.Transpose.ROTATE_90,
+        self,
-}
+        message: str,
+        *,
+        stderr: str | None = None,
+        backend: str | None = None,
+    ):
+        super().__init__(message)
+        self.stderr = stderr
+        self.backend = backend
+
+
+async def _run_preview_process(
+    filepath, quality: int, maxsize: int, maxzoom: float
+) -> 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)
+
+
+DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
+
+
+def is_previewable_path(path) -> bool:
+    suffix = path.suffix.lower()
+    if suffix in DOC_PREVIEW_SUFFIXES:
+        return True
+    mime_type, _ = mimetypes.guess_type(path.name)
+    if not mime_type:
+        return False
+    return mime_type.startswith("image/") or mime_type.startswith("video/")
 
 
 @bp.get("/<path:path>")
@@ -97,7 +338,14 @@ async def preview(req, path):
     quality = int(req.args.get("q", 60))
     rel = PurePosixPath(sanitize(unquote(path)))
     filepath = config.config.path / rel
+    try:
         stat = filepath.lstat()
+    except FileNotFoundError:
+        raise NotFound() from None
+
+    if not is_previewable_path(filepath):
+        return empty(415)
+
     etag = config.derived_secret(
         "preview", rel, stat.st_mtime_ns, quality, maxsize, maxzoom
     ).hex()
@@ -112,25 +360,48 @@ async def preview(req, path):
         logger.debug(f"Preview cache hit: {rel}")
         return raw(cached.body, headers=cached.headers)
 
-    if not filepath.is_file():
-        raise NotFound("File not found")
-
     # Generate preview
-    img = await asyncio.get_event_loop().run_in_executor(
+    try:
-        req.app.ctx.threadexec, dispatch, filepath, quality, maxsize, maxzoom
+        img, preview_resp = await _run_preview_process(
+            filepath, quality, maxsize, maxzoom
         )
+    except PreviewTimeout:
+        return empty(504)
+    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)
+    if preview_resp and preview_resp.backend:
+        if preview_resp.timings:
+            timing_detail = "/".join(
+                str(int(round(value))) for value in preview_resp.timings
+            )
+            req.ctx._log_extra = f"{preview_resp.backend} {timing_detail} ➛"
+        else:
+            req.ctx._log_extra = preview_resp.backend
     if not img:
         # Preview generation failed, redirect to the file itself
         return redirect(f"/files/{path}", status=303)
 
     # Build headers and cache the full response
+    preview_mime = (
+        preview_resp.mime
+        if preview_resp is not None and preview_resp.mime is not None
+        else "image/avif"
+    )
     savename = PurePosixPath(filepath.name).with_suffix(".avif")
     headers = {
         "etag": etag,
         "last-modified": format_date_time(stat.st_mtime),
         "cache-control": "max-age=604800, immutable"
         + ("" if config.config.public else ", private"),
-        "content-type": "image/avif",
+        "content-type": preview_mime,
         "content-disposition": f"inline; filename*=UTF-8''{urllib.parse.quote(savename.as_posix())}",
     }
     _preview_cache.set(etag, CachedPreview(headers=headers, body=img))
@@ -139,52 +410,50 @@ async def preview(req, path):
 
 
 def dispatch(path, quality, maxsize, maxzoom):
+    backend = "unknown"
     try:
-        if path.suffix.lower() in (".pdf", ".xps", ".epub", ".mobi"):
+        if path.suffix.lower() in DOC_PREVIEW_SUFFIXES:
+            backend = "pdf"
             return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
-        type, _ = mimetypes.guess_type(path.name)
+        mime_type, _ = mimetypes.guess_type(path.name)
-        if type and type.startswith("video/"):
+        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:
-        logger.warning(f"Cannot generate preview for {path}: {e}")
+        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
     except Exception as e:
-        logger.exception(f"Error generating preview for {path}: {e}")
+        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):
-    t_load = perf_counter()
+    return process_image_pyvips(path, maxsize=maxsize, quality=quality)
-    with Image.open(path) as img:
-        # Force decode to include I/O in load timing
-        img.load()
-        t_proc = perf_counter()
-        # Resize
-        w, h = img.size
-        img.thumbnail((min(w, maxsize), min(h, maxsize)))
-        # Transpose pixels according to EXIF Orientation
-        orientation = img.getexif().get(274, 1)
-        if orientation in EXIF_ORI:
-            img = img.transpose(EXIF_ORI[orientation])
-        # Save as AVIF
-        imgdata = io.BytesIO()
-        t_save = perf_counter()
-        img.save(imgdata, format="avif", quality=quality, speed=10, max_threads=1)
 
-    t_end = perf_counter()
-    ret = imgdata.getvalue()
 
-    load_ms = (t_proc - t_load) * 1000
+def process_image_pyvips(path, *, maxsize, quality):
-    proc_ms = (t_save - t_proc) * 1000
+    t_start = perf_counter()
-    save_ms = (t_end - t_save) * 1000
+    img = pyvips.Image.new_from_file(str(path), access="sequential")
-    logger.debug(
+    img = img.autorot()
-        "Preview image %s: load=%.1fms process=%.1fms save=%.1fms",
+    scale = min(maxsize / img.width, maxsize / img.height, 1.0)
-        path.name,
+    if scale < 1.0:
-        load_ms,
+        img = img.resize(scale)
-        proc_ms,
+    ret = img.write_to_buffer(
-        save_ms,
+        ".avif",
+        Q=quality,
+        effort=AVIF_FAST_EFFORT,
+        strip=True,
     )
+    t_end = perf_counter()
 
-    return ret
+    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):
@@ -198,16 +467,24 @@ def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
     t_load_end = perf_counter()
 
     t_save_start = perf_counter()
-    ret = pix.pil_tobytes(format="avif", quality=quality, speed=10, max_threads=1)
+    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()
 
-    logger.debug(
+    return ret, PreviewResponse(
-        "Preview pdf %s: load+render=%.1fms save=%.1fms",
+        ok=True,
-        path.name,
+        mime="image/avif",
-        (t_load_end - t_load_start) * 1000,
+        backend=backend,
-        (t_save_end - t_save_start) * 1000,
+        timings=[
+            round((t_load_end - t_load_start) * 1000, 1),
+            round((t_save_end - t_save_start) * 1000, 1),
+        ],
     )
-    return ret
 
 
 def process_video(path, *, maxsize, quality):
@@ -220,7 +497,13 @@ def process_video(path, *, maxsize, quality):
     t_save_start = t_load_start
     t_save_end = t_load_start
     with (
-        av.open(str(path)) as icontainer,
+        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]
@@ -312,14 +595,17 @@ def process_video(path, *, maxsize, quality):
         ocontainer.mux(ostream.encode(None))  # Flush the stream
         t_save_end = perf_counter()
 
-    # Capture frame dimensions before cleanup
+    # Capture result before cleanup
     ret = imgdata.getvalue()
-    logger.debug(
+    resp = PreviewResponse(
-        "Preview video %s: load+decode=%.1fms save=%.1fms",
+        ok=True,
-        path.name,
+        mime="image/avif",
-        (t_load_end - t_load_start) * 1000,
+        backend="video",
-        (t_save_end - t_save_start) * 1000,
+        timings=[
+            round((t_load_end - t_load_start) * 1000, 1),
+            round((t_save_end - t_save_start) * 1000, 1),
+        ],
     )
     del imgdata, istream, ostream, icc, occ, frame
     gc.collect()
-    return ret
+    return ret, resp

cista/preview_worker.py

@@ -0,0 +1,116 @@
+"""Preview generation worker subprocess.
+
+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.
+
+Framed response format:
+    (blake3(packet))(uint32 json size)(uint32 payload size)(json)(binary payload)
+where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
+"""
+
+import logging
+import contextlib
+import io
+import struct
+import sys
+from pathlib import Path
+
+import msgspec
+from blake3 import blake3
+
+
+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
+
+
+_enc = msgspec.json.Encoder()
+_dec_req = msgspec.json.Decoder(PreviewRequest)
+
+
+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 _run_once() -> None:
+    if len(sys.argv) != 5:
+        sys.stderr.write(f"Usage: {sys.argv[0]} <path> <quality> <maxsize> <maxzoom>\n")
+        sys.exit(1)
+
+    from cista.preview import dispatch
+
+    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:
+    from cista.preview import dispatch
+
+    while True:
+        line = sys.stdin.buffer.readline()
+        if not line:
+            return
+        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
+                )
+            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:
+            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)
+    if len(sys.argv) > 1:
+        _run_once()
+        return
+    _run_loop()
+
+
+if __name__ == "__main__":
+    main()

cista/protocol.py

@@ -12,7 +12,6 @@ from cista.util import filename
 
 ## Control commands
 
-
 class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower):
     def __call__(self):
         raise NotImplementedError
@@ -118,19 +117,9 @@ class Cp(ControlBase):
 ControlTypes = MkDir | Rename | Rm | Mv | Cp
 
 
-## File uploads and downloads
-
-
-class FileRange(msgspec.Struct):
-    name: str
-    size: int
-    start: int
-    end: int
-
-
 class StatusMsg(msgspec.Struct):
     status: str
-    req: FileRange
+    req: Any
 
 
 class ErrorMsg(msgspec.Struct):

cista/sanic_logging.py

@@ -0,0 +1,257 @@
+"""Custom access logging middleware for Sanic."""
+
+import logging
+import sys
+import unicodedata
+from ipaddress import IPv6Address
+
+logger = logging.getLogger("cista.access")
+
+_RESET = "\033[0m"
+_STATUS_INFO = "\033[32m"      # 1xx (green)
+_STATUS_OK = "\033[1;92m"      # 2xx (bright green)
+_STATUS_REDIRECT = "\033[32m"  # 3xx (green)
+_STATUS_CLIENT_ERR = "\033[0;31m"   # 4xx (red)
+_STATUS_SERVER_ERR = "\033[1;91m"   # 5xx (bold bright red)
+_METHOD_READ = "\033[0;34m"    # GET, HEAD, OPTIONS (blue)
+_METHOD_WRITE = "\033[1;94m"   # POST, PUT, DELETE, PATCH (bold bright blue)
+_HOST = "\033[38;5;242m"       # hostname (dark grey)
+_PATH = "\033[38;5;250m"       # path (light grey)
+_TIMING = "\033[38;5;242m"     # timing (dark grey)
+_WS_OPEN = "\033[1;93m"        # WebSocket connect (bold bright yellow)
+_WS_CLOSE = "\033[33m"         # WebSocket disconnect (yellow)
+_WS_STATUS = "\033[38;5;250m"  # WebSocket close status (normal white)
+
+
+def format_ipv6_network(ip: str) -> str:
+    """Format IPv6 address to show only network part (first 64 bits)."""
+    try:
+        ip = ip.strip("[]")
+        if "%" in ip:
+            ip = ip.split("%")[0]
+        addr = IPv6Address(ip)
+        if addr.is_loopback:
+            return "::1"
+        if addr.is_unspecified:
+            return "::"
+        if addr.ipv4_mapped:
+            return str(addr.ipv4_mapped)
+        if addr.is_link_local:
+            return str(addr)
+        network_int = int(addr) >> 64
+        groups = []
+        for _ in range(4):
+            groups.insert(0, format(network_int & 0xFFFF, "x"))
+            network_int >>= 16
+        result = ":".join(groups) + "::"
+        return str(IPv6Address(result + "0")).removesuffix("::")
+    except Exception:
+        return ip
+
+
+def format_client_ip(ip: str) -> str:
+    """Format client IP, compressing IPv6 to network part only."""
+    if not ip or ip == "-":
+        return "-"
+    stripped = ip.strip("[]")
+    if ":" in stripped:
+        return format_ipv6_network(stripped)
+    return stripped
+
+
+def status_color(status: int) -> str:
+    if status < 200:
+        return _STATUS_INFO
+    if status < 300:
+        return _STATUS_OK
+    if status < 400:
+        return _STATUS_REDIRECT
+    if status < 500:
+        return _STATUS_CLIENT_ERR
+    return _STATUS_SERVER_ERR
+
+
+def method_color(method: str) -> str:
+    if method in ("GET", "HEAD", "OPTIONS"):
+        return _METHOD_READ
+    return _METHOD_WRITE
+
+
+def format_duration_ms(duration_ms: float) -> str:
+    rounded_ms = round(duration_ms)
+    if rounded_ms < 2000:
+        return f"{rounded_ms}ms"
+    total_s = round(duration_ms / 1000)
+    if total_s < 60:
+        return f"{total_s}s"
+    if total_s <= 3600:
+        minutes, seconds = divmod(total_s, 60)
+        return f"{minutes}m{seconds}s"
+    hours, remainder = divmod(total_s, 3600)
+    minutes = round(remainder / 60)
+    if minutes == 60:
+        hours += 1
+        minutes = 0
+    return f"{hours}h{minutes}m"
+
+
+def _display_width(text: str) -> int:
+    width = 0
+    for char in text:
+        width += 2 if unicodedata.east_asian_width(char) in {"F", "W"} else 1
+    return width
+
+
+def _format_left(label: str) -> str:
+    return label[:19].ljust(19)
+
+
+def _format_method_label(label: str, *, color: str | None = None) -> str:
+    color_value = _METHOD_WRITE if color is None else color
+    padding = max(0, 7 - _display_width(label))
+    return f"{color_value}{label}{' ' * padding}{_RESET}"
+
+
+def format_access_log(
+    client: str, status: int, method: str, host: str, path: str, duration_ms: float,
+    extra: str | None = None,
+) -> str:
+    ip = _format_left(format_client_ip(client))
+    status_str = f"{status_color(status)}{str(status).rjust(3)}{_RESET}"
+    method_str = _format_method_label(method, color=method_color(method))
+    host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
+    timing_str = f"{_TIMING}{format_duration_ms(duration_ms)}{_RESET}"
+    extra_str = f" {_TIMING}{extra}{_RESET}" if extra else ""
+    return f"{ip} {status_str} {method_str} {host_str}{path_str}{extra_str} {timing_str}"
+
+
+_ws_counter = 1
+
+
+def _next_ws_id() -> int:
+    global _ws_counter
+    ws_id = _ws_counter
+    _ws_counter += 1
+    return ws_id
+
+
+def _format_ws_id(ws_id: int, *, bright: bool = False) -> str:
+    value = str(ws_id) if ws_id >= 100 else f"{ws_id:02d}"
+    color = _WS_OPEN if bright else _WS_CLOSE
+    return f"{color}{value.rjust(3)}{_RESET}"
+
+
+def log_ws_open(request, extra: str | None = None) -> int:
+    """Log WebSocket connection open. Returns connection ID for use in log_ws_close."""
+    ws_id = _next_ws_id()
+
+    client = request.client_ip or "-"
+    host = request.host or "-"
+    path = request.path
+    origin = request.headers.get("origin")
+
+    ip = _format_left(format_client_ip(client))
+    id_str = _format_ws_id(ws_id, bright=True)
+
+    origin_host = origin.split("://", 1)[-1] if origin else None
+    show_origin = origin_host and origin_host != host
+
+    method_str = _format_method_label("🔌", color=_WS_OPEN)
+    host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
+    origin_str = f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else ""
+    extra_str = f" {_TIMING}{extra}{_RESET}" if extra else ""
+
+    logger.info(
+        "%s %s %s %s%s%s",
+        ip,
+        id_str,
+        method_str,
+        host_str,
+        path_str,
+        origin_str + extra_str,
+    )
+    return ws_id
+
+
+WS_CLOSE_CODES = {
+    1000: "ok",
+    1001: "going away",
+    1002: "protocol error",
+    1003: "unsupported",
+    1005: "no status",
+    1006: "abnormal",
+    1007: "invalid data",
+    1008: "policy violation",
+    1009: "too large",
+    1010: "extension required",
+    1011: "server error",
+    1012: "restarting",
+    1013: "try again",
+    1014: "bad gateway",
+    1015: "tls error",
+}
+
+
+def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None:
+    """Log WebSocket connection close with duration and status."""
+    id_str = _format_ws_id(ws_id)
+    timing = format_duration_ms(duration * 1000)
+
+    if close_code is None:
+        code = "----"
+        status = "unknown"
+    else:
+        code = str(close_code)
+        status = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
+
+    method_str = _format_method_label("closed", color=_TIMING)
+    status_str = f"{_WS_STATUS}{code} {status}{_RESET}"
+    timing_str = f"{_TIMING}{timing}{_RESET}"
+
+    logger.info("%s %s %s %s %s", " " * 19, id_str, method_str, status_str, timing_str)
+
+
+def configure_access_logging() -> None:
+    """Configure the cista.access logger to output to stderr."""
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(logging.Formatter("%(message)s"))
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+
+
+_LEVEL_EMOJI = {
+    logging.DEBUG: "🔍",
+    logging.INFO: "ℹ️",
+    logging.WARNING: "⚠️",
+    logging.ERROR: "🛑",
+    logging.CRITICAL: "🛑",
+}
+
+
+class _EmojiFormatter(logging.Formatter):
+    """Compact formatter: emoji + message, no timestamp/level text/logger name."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        emoji = _LEVEL_EMOJI.get(record.levelno, "▪️")
+        sep = "  " if record.levelno in (logging.INFO, logging.WARNING) else " "
+        return f"{emoji}{sep}{record.getMessage()}"
+
+
+def configure_main_logging() -> None:
+    """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().
+    """
+    from sanic.log import LOGGING_CONFIG_DEFAULTS
+
+    LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
+        "class": "cista.sanic_logging._EmojiFormatter",
+    }
+    # 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())

cista/serve.py

@@ -27,7 +27,7 @@ def run(*, dev=False):
         motd=False,
         dev=dev,
         auto_reload=dev,
-        access_log=True,
+        access_log=False,
     )  # type: ignore
     if dev:
         Sanic.serve()

cista/util/apphelpers.py

@@ -1,3 +1,4 @@
+import time
 from functools import wraps
 
 import msgspec
@@ -8,6 +9,7 @@ from sanic.response import raw, redirect
 
 from cista import auth
 from cista.protocol import ErrorMsg
+from cista.sanic_logging import log_ws_close, log_ws_open
 
 
 def asend(ws, msg):
@@ -54,6 +56,10 @@ def websocket_wrapper(handler):
 
     @wraps(handler)
     async def wrapper(request, ws, *args, **kwargs):
+        username = getattr(request.ctx, "username", None)
+        extra = username if username else None
+        start = time.perf_counter()
+        ws_id = log_ws_open(request, extra=extra)
         try:
             await auth.verify(request)
             await handler(request, ws, *args, **kwargs)
@@ -67,5 +73,19 @@ def websocket_wrapper(handler):
             if not getattr(e, "quiet", False) or code == 500:
                 logger.exception(f"{code} {e!r}")
             raise
+        finally:
+            duration = time.perf_counter() - start
+            close_code = None
+            try:
+                p = ws.ws_proto
+                if p.close_rcvd is not None:
+                    close_code = p.close_rcvd.code
+                elif p.close_sent is not None:
+                    close_code = p.close_sent.code
+                elif getattr(p, "close_code", None) is not None:
+                    close_code = p.close_code
+            except AttributeError:
+                pass
+            log_ws_close(ws_id, close_code, duration)
 
     return wrapper

frontend/src/components/MediaPreview.vue

@@ -1,9 +1,13 @@
 <template>
-  <img v-if=preview() :src="`${doc.previewurl}?${quality}&t=${doc.mtime}`" alt="">
+  <div v-if=showProgress() class="preview-progress" aria-label="Preview pending">
+    <SpinnerIcon />
+  </div>
+  <img v-else-if="previewSrc && !video() && !audio()" :src="previewSrc" alt="">
   <img v-else-if=doc.img :src=doc.url alt="">
   <span v-else-if=doc.dir class="folder icon"></span>
-  <div v-else-if=video() class="video-container">
+  <div v-else-if=video() class="video-container" :class="{ pending: !doc.complete }">
-    <video ref=vid :src=doc.url :poster=poster preload=none @play=onplay @pause=onpaused @ended=next @seeking=media!.play()></video>
+    <video v-if=doc.complete ref=vid :src=doc.url :poster=previewSrc preload=none @play=onplay @pause=onpaused @ended=next @seeking=media!.play()></video>
+    <video v-else ref=vid :src=doc.url preload=none @play=onplay @pause=onpaused @ended=next @seeking=media!.play()></video>
     <div class="play-overlay"><PlayIcon /></div>
   </div>
   <div v-else-if=audio() class="audio icon">
@@ -16,16 +20,16 @@
 <script setup lang=ts>
 import { computed, ref } from 'vue'
 import type { Doc } from '@/repositories/Document'
-import { Play as PlayIcon } from '@/assets/svg'
+import { Play as PlayIcon, Spinner as SpinnerIcon } from '@/assets/svg'
 
 const aud = ref<HTMLAudioElement | null>(null)
 const vid = ref<HTMLVideoElement | null>(null)
 const media = computed(() => aud.value || vid.value)
-const poster = computed(() => `${props.doc.previewurl}?${props.quality}&t=${props.doc.mtime}`)
 const props = defineProps<{
   doc: Doc
   quality: string
 }>()
+const previewSrc = computed(() => props.doc.previewurl ? `${props.doc.previewurl}?${props.quality}&t=${props.doc.mtime}` : '')
 
 const onplay = () => {
   if (!media.value) return
@@ -37,6 +41,13 @@ const onpaused = () => {
   media.value.controls = false
   media.value.removeAttribute('data-playing')
 }
+const applyPoster = (el: HTMLVideoElement) => {
+  if (props.doc.complete) {
+    el.poster = previewSrc.value
+  } else {
+    el.removeAttribute('poster')
+  }
+}
 let fscurrent: HTMLVideoElement | null = null
 const next = () => {
   if (!media.value) return
@@ -70,7 +81,7 @@ const next = () => {
         if (!elem.paused) fscurrent.play()
         fscurrent = null
         elem.src = props.doc.url
-        elem.poster = poster.value
+        applyPoster(elem)
         onpaused()
       }, {once: true})
     }
@@ -104,6 +115,7 @@ defineExpose({
 const video = () => ['mkv', 'mp4', 'webm', 'mov', 'avi'].includes(props.doc.ext)
 const audio = () => ['mp3', 'flac', 'ogg', 'aac'].includes(props.doc.ext)
 const archive = () => ['zip', 'tar', 'gz', 'bz2', 'xz', '7z', 'rar'].includes(props.doc.ext)
+const showProgress = () => !props.doc.complete && (preview() || props.doc.img)
 const preview = () => (
   ['bmp', 'ico', 'tif', 'tiff', 'heic', 'heif', 'pdf', 'epub', 'mobi'].includes(props.doc.ext) ||
   props.doc.size > 500000 &&
@@ -120,6 +132,29 @@ img, embed, .icon, audio, video {
   max-height: 100%;
   border-radius: calc(.5em / 8);
 }
+.preview-progress {
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  min-width: 50%;
+  max-width: 100%;
+  max-height: 100%;
+  aspect-ratio: 1;
+}
+.preview-progress :deep(svg) {
+  width: 4.5em;
+  height: 4.5em;
+  opacity: 0.8;
+  animation: media-preview-spin 0.9s linear infinite;
+}
+@keyframes media-preview-spin {
+  from {
+    transform: rotate(0deg);
+  }
+  to {
+    transform: rotate(360deg);
+  }
+}
 .folder::before {
   content: '📁';
 }
@@ -175,9 +210,14 @@ img::before {
   align-items: center;
   justify-content: center;
   min-width: 50%;
+  min-height: 6em;
+  aspect-ratio: 16 / 9;
   max-width: 100%;
   max-height: 100%;
 }
+.video-container.pending {
+  background: color-mix(in srgb, var(--header-bg) 55%, transparent);
+}
 .video-container video {
   width: 100%;
   height: 100%;

frontend/src/components/UploadButton.vue

@@ -8,12 +8,11 @@
 </template>
 
 <script setup lang="ts">
-import { connect, uploadUrl } from '@/repositories/WS';
 import { useMainStore } from '@/stores/main'
 import { getDocuments } from '@/stores/documentStore'
 import { Doc } from '@/repositories/Document'
 import { collator } from '@/utils';
-import { onMounted, onUnmounted, reactive, ref } from 'vue'
+import { onMounted, onUnmounted, ref } from 'vue'
 import { useRouter } from 'vue-router'
 
 const router = useRouter()
@@ -29,6 +28,22 @@ type CloudFile = {
   cloudName: string
   cloudPos: number
 }
+
+type UploadRange = {
+  name: string
+  size: number
+  start: number
+  end: number
+}
+
+type InflightBlock = {
+  name: string
+  start: number
+  end: number
+  startedAt: number
+}
+
+const UPLOAD_BLOCK_SIZE = 16 << 20  // 16 MiB
 function pasteHandler(event: ClipboardEvent) {
   const items = Array.from(event.clipboardData?.items ?? [])
   const infiles = [] as File[]
@@ -46,7 +61,8 @@ function pasteHandler(event: ClipboardEvent) {
   if (infiles.length || dirs.length) {
     event.preventDefault()
     uploadFiles(infiles)
-    for (const entry of dirs) pasteDirectory(entry, `${props.path!.join('/')}/${entry.name}`)
+    const base = props.path!.join('/')
+    for (const entry of dirs) pasteDirectory(entry, `${base ? `${base}/` : ''}${entry.name}`)
   }
 }
 const pasteDirectory = async (entry: FileSystemDirectoryEntry, loc: string) => {
@@ -82,7 +98,7 @@ const uploadFiles = (infiles: File[]) => {
     if (!folderName && file.webkitRelativePath) folderName = relPath.split('/')[0] ?? ''
     files.push({
       file,
-      cloudName: loc + '/' + relPath,
+      cloudName: `${loc ? `${loc}/` : ''}${relPath}`,
       cloudPos: 0,
     })
   }
@@ -130,7 +146,12 @@ const uploadCloudFiles = (files: CloudFile[]) => {
 }
 
 const cancelUploads = () => {
+  uploadRunId += 1
   upqueue = []
+  blockQueue = []
+  inflightBlocks.clear()
+  uploadedBytes.clear()
+  store.uprogress.status = 'idle'
   statReset()
 }
 
@@ -151,27 +172,100 @@ const uprogress_init = {
   status: 'idle',
 }
 store.uprogress = {...uprogress_init}
+// Track uploaded bytes for each file to handle out-of-order uploads
+const uploadedBytes = new Map<string, Set<number>>()
+const inflightBlocks = new Map<string, InflightBlock>()
+let smoothedBlockMs = 1500
+let lastProgressTick = Date.now()
+let lastVisualUploaded = 0
+
+const inflightKey = (name: string, start: number) => `${name}:${start}`
+
+const completedUploadedBytes = (name: string, size: number) => {
+  const uploaded = uploadedBytes.get(name)
+  if (!uploaded) return 0
+  const blockSize = UPLOAD_BLOCK_SIZE
+  let total = 0
+  for (let i = 0; i < size; i += blockSize) {
+    if (uploaded.has(i)) total += Math.min(blockSize, size - i)
+  }
+  return total
+}
+
+const simulatedInflightBytes = (name: string, now: number) => {
+  let total = 0
+  for (const block of inflightBlocks.values()) {
+    if (block.name !== name) continue
+    const size = block.end - block.start
+    const elapsed = Math.max(0, now - block.startedAt)
+    const fraction = Math.min(0.98, elapsed / Math.max(200, smoothedBlockMs))
+    total += size * fraction
+  }
+  return total
+}
+
+const refreshProgress = (now: number) => {
+  const name = store.uprogress.filename
+  const size = store.uprogress.filesize
+  if (!name || !size) {
+    lastProgressTick = now
+    return 0
+  }
+
+  const completed = completedUploadedBytes(name, size)
+  const estimated = simulatedInflightBytes(name, now)
+  const visualUploaded = Math.min(size, Math.round(completed + estimated))
+  const delta = Math.max(0, visualUploaded - lastVisualUploaded)
+  const dt = Math.max(1, now - lastProgressTick)
+
+  store.uprogress.filepos = visualUploaded
+  store.uprogress.xfer = store.uprogress.filestart + visualUploaded
+
+  if (delta > 0) {
+    store.uprogress.statbytes += delta
+    store.uprogress.statdur += dt
+    store.uprogress.tlast = now
+  }
+
+  lastVisualUploaded = visualUploaded
+  lastProgressTick = now
+  return delta
+}
+
 setInterval(() => {
-  if (Date.now() - store.uprogress.tlast > 3000) {
+  const now = Date.now()
-    // Reset
+  const delta = refreshProgress(now)
+  if (delta > 0) return
+  if (now - store.uprogress.tlast > 3000) {
     store.uprogress.statbytes = 0
     store.uprogress.statdur = 1
   } else {
-    // Running average by decay
+    store.uprogress.statbytes *= .95
-    store.uprogress.statbytes *= .9
+    store.uprogress.statdur *= .95
-    store.uprogress.statdur *= .9
   }
 }, 100)
-const statUpdate = ({name, size, start, end}: {name: string, size: number, start: number, end: number}) => {
+
+const statUpdate = ({name, size, start, end}: UploadRange) => {
   if (name !== store.uprogress.filename) return  // If stats have been reset
-  const now = Date.now()
+
-  store.uprogress.xfer = store.uprogress.filestart + end
+  // Track which bytes have been uploaded (using start to end range)
-  store.uprogress.filepos = end
+  if (!uploadedBytes.has(name)) uploadedBytes.set(name, new Set())
-  store.uprogress.statbytes += end - start
+  const uploaded = uploadedBytes.get(name)!
-  store.uprogress.statdur += now - store.uprogress.tlast
+  const blockSize = UPLOAD_BLOCK_SIZE
-  store.uprogress.tlast = now
+
-  // File finished?
+  // Mark all bytes in this block as uploaded
-  if (end === size) {
+  for (let i = start; i < end; i += blockSize) {
+    uploaded.add(i)
+  }
+  refreshProgress(Date.now())
+
+  // Check if file is fully uploaded by examining the block queue
+  const currentUpload = blockQueue[0]
+  if (!currentUpload) return
+
+  if (currentUpload.file.cloudName === name && currentUpload.completed >= currentUpload.blocks.length) {
+    // All blocks for this file have been uploaded
+    uploadedBytes.delete(name)  // Clean up tracking
     store.uprogress.filestart += size
     statNextFile()
     if (++store.uprogress.fileidx >= store.uprogress.filecount) statReset()
@@ -183,11 +277,15 @@ const statNextFile = () => {
   store.uprogress.filepos = 0
   store.uprogress.filesize = f.file.size
   store.uprogress.filename = f.cloudName
+  lastVisualUploaded = 0
+  lastProgressTick = Date.now()
 }
 const statReset = () => {
   Object.assign(store.uprogress, uprogress_init)
   store.uprogress.t0 = Date.now()
   store.uprogress.tlast = store.uprogress.t0 + 1
+  lastVisualUploaded = 0
+  lastProgressTick = store.uprogress.t0
 }
 const statsAdd = (f: CloudFile[]) => {
   if (store.uprogress.files.length === 0) statReset()
@@ -197,69 +295,201 @@ const statsAdd = (f: CloudFile[]) => {
   statNextFile()
 }
 let upqueue = [] as CloudFile[]
+const MAX_PARALLEL_REQUESTS = 4
+const RETRY_DELAY_MS = 400
 
-// TODO: Rewrite as WebSocket class
+// Helper function to get upload blocks for a file, prioritizing final 4 blocks if file >= 32 MiB
-const WSCreate = async () => await new Promise<WebSocket>(resolve => {
+const getUploadBlocks = (file: CloudFile): {start: number, end: number}[] => {
-  const ws = connect(uploadUrl, {
+  const BLOCK_SIZE = UPLOAD_BLOCK_SIZE
-    open(ev: Event) { resolve(ws) },
+  const MIN_SIZE_FOR_REORDER = 32 * BLOCK_SIZE  // 32 MiB = 33554432 bytes
-    error(ev: Event) {
+  const FINAL_BLOCKS_COUNT = 2
-      console.error('Upload socket error', ev)
+
-      store.error = 'Upload socket error'
+  const fileSize = file.file.size
-    },
+  const blocks: {start: number, end: number}[] = []
-    message(ev: MessageEvent) {
+
-      const res = JSON.parse(ev!.data)
+  if (fileSize >= MIN_SIZE_FOR_REORDER) {
-      if ('error' in res) {
+    // File is large enough, prioritize final blocks
-        console.error('Upload socket error', res.error)
+    const finalBlocksStart = fileSize - (FINAL_BLOCKS_COUNT * BLOCK_SIZE)
-        store.error = res.error.message
+
-        return
+    // Add final blocks first
+    for (let i = 0; i < FINAL_BLOCKS_COUNT; i++) {
+      const start = finalBlocksStart + (i * BLOCK_SIZE)
+      const end = Math.min(start + BLOCK_SIZE, fileSize)
+      blocks.push({start, end})
     }
-      if (res.status === 'ack') {
+
-        statUpdate(res.req)
+    // Add remaining blocks from beginning
-      } else console.log('Unknown upload response', res)
+    for (let start = 0; start < finalBlocksStart; start += BLOCK_SIZE) {
-    },
+      const end = Math.min(start + BLOCK_SIZE, finalBlocksStart)
-  })
+      blocks.push({start, end})
-  // @ts-ignore
-  ws.sendMsg = (msg: any) => ws.send(JSON.stringify(msg))
-  // @ts-ignore
-  ws.sendData = async (data: any) => {
-    // Wait until the WS is ready to send another message
-    store.uprogress.status = "uploading"
-    await new Promise(resolve => {
-      const t = setInterval(() => {
-        if (ws.bufferedAmount > 1<<20) return
-        resolve(undefined)
-        clearInterval(t)
-      }, 1)
-    })
-    store.uprogress.status = "processing"
-    ws.send(data)
     }
-})
+  } else {
-const worker = async () => {
+    // File is smaller, use sequential upload
-  const ws = await WSCreate()
+    for (let start = 0; start < fileSize; start += BLOCK_SIZE) {
+      const end = Math.min(start + BLOCK_SIZE, fileSize)
+      blocks.push({start, end})
+    }
+  }
+
+  return blocks
+}
+
+type BlockUpload = {
+  file: CloudFile
+  blocks: {start: number, end: number}[]
+  nextIndex: number
+  completed: number
+  runId: number
+}
+
+let blockQueue = [] as BlockUpload[]
+let workerRunning = false
+let uploadRunId = 0
+
+const enqueuePendingUploads = () => {
   while (upqueue.length) {
-    const f = upqueue[0]!
+    const file = upqueue.shift()!
-    const start = f.cloudPos
+    const blocks = getUploadBlocks(file)
-    const end = Math.min(f.file.size, start + (1<<20))
+    blockQueue.push({ file, blocks, nextIndex: 0, completed: 0, runId: uploadRunId })
-    const control = { name: f.cloudName, size: f.file.size, start, end }
-    const data = f.file.slice(start, end)
-    f.cloudPos = end
-    // Note: files may get modified during I/O
-    // @ts-ignore FIXME proper WebSocket class, avoid attaching functions to WebSocket object
-    ws.sendMsg(control)
-    // @ts-ignore
-    await ws.sendData(data)
-    if (f.cloudPos === f.file.size) upqueue.shift()
   }
-  if (upqueue.length) startWorker()
+}
-  store.uprogress.status = "idle"
+
+const uploadUrlForFile = (cloudName: string) => {
+  const normalized = cloudName.replace(/^\/+/, '')
+  const encoded = normalized.split('/').map(encodeURIComponent).join('/')
+  return `/files/${encoded}`
+}
+
+const uploadBlock = async (upload: BlockUpload, block: {start: number, end: number}) => {
+  const body = upload.file.file.slice(block.start, block.end)
+  const range = `bytes ${block.start}-${block.end - 1}/${upload.file.file.size}`
+  const fallbackReq = {
+    name: upload.file.cloudName,
+    size: upload.file.file.size,
+    start: block.start,
+    end: block.end,
+  }
+  let attempt = 0
+
+  while (true) {
+    attempt += 1
+    if (upload.runId !== uploadRunId) throw new Error('Upload cancelled')
+    try {
+      const res = await fetch(uploadUrlForFile(upload.file.cloudName), {
+        method: 'PUT',
+        headers: {
+          'Content-Type': 'application/octet-stream',
+          'Content-Range': range,
+        },
+        body,
+      })
+      if (!res.ok) {
+        const message = await res.text().catch(() => '')
+        const retryable = res.status >= 500 || res.status === 408 || res.status === 429
+        if (!retryable) throw new Error(message || `HTTP ${res.status}`)
+        await new Promise(resolve => setTimeout(resolve, RETRY_DELAY_MS))
+        continue
+      }
+      const payload = await res.json().catch(() => null)
+      return payload?.status === 'ack' && payload.req ? payload.req : fallbackReq
+    } catch (err: any) {
+      const message = err instanceof Error ? err.message : String(err)
+      if (message === 'Upload cancelled') throw err
+      if (upload.runId !== uploadRunId) throw new Error('Upload cancelled')
+      if (attempt % 10 === 0) {
+        console.warn(`Upload retry ${attempt} for ${upload.file.cloudName}: ${message}`)
+      }
+      await new Promise(resolve => setTimeout(resolve, RETRY_DELAY_MS))
+    }
+  }
+}
+
+const startInflightBlock = (name: string, block: {start: number, end: number}) => {
+  inflightBlocks.set(inflightKey(name, block.start), {
+    name,
+    start: block.start,
+    end: block.end,
+    startedAt: Date.now(),
+  })
+}
+
+const finishInflightBlock = (name: string, block: {start: number, end: number}) => {
+  const key = inflightKey(name, block.start)
+  const info = inflightBlocks.get(key)
+  if (!info) return
+  const elapsed = Math.max(1, Date.now() - info.startedAt)
+  smoothedBlockMs = smoothedBlockMs * 0.85 + elapsed * 0.15
+  inflightBlocks.delete(key)
+}
+
+const worker = async (runId: number) => {
+  try {
+    while (runId === uploadRunId) {
+      enqueuePendingUploads()
+      if (!blockQueue.length) break
+
+      const upload = blockQueue[0]!
+      const inflight = new Set<Promise<void>>()
+
+      while (runId === uploadRunId && upload.completed < upload.blocks.length) {
+        while (
+          runId === uploadRunId
+          && upload.nextIndex < upload.blocks.length
+          && inflight.size < MAX_PARALLEL_REQUESTS
+        ) {
+          const block = upload.blocks[upload.nextIndex++]!
+          store.uprogress.status = 'uploading'
+          startInflightBlock(upload.file.cloudName, block)
+          let task: Promise<void>
+          task = uploadBlock(upload, block)
+            .then(req => {
+              finishInflightBlock(upload.file.cloudName, block)
+              upload.completed += 1
+              statUpdate(req)
+            })
+            .catch(err => {
+              finishInflightBlock(upload.file.cloudName, block)
+              throw err
+            })
+            .finally(() => {
+              inflight.delete(task)
+            })
+          inflight.add(task)
+        }
+
+        if (!inflight.size) break
+        await Promise.race(inflight)
+      }
+
+      if (runId !== uploadRunId) return
+
+      if (upload.completed >= upload.blocks.length) {
+        blockQueue.shift()
+      } else {
+        break
+      }
+    }
+  } catch (err: any) {
+    if (runId !== uploadRunId) return
+    console.error('Upload error', err)
+    store.error = err?.message || 'Upload failed'
+    uploadRunId += 1
+    upqueue = []
+    blockQueue = []
+    inflightBlocks.clear()
+  } finally {
+    store.uprogress.status = 'idle'
     workerRunning = false
+    if (upqueue.length) startWorker()
   }
-let workerRunning: any = false
+}
+
 const startWorker = () => {
-  if (workerRunning === false) workerRunning = setTimeout(() => {
+  if (workerRunning) return
   workerRunning = true
-    worker()
+  const runId = uploadRunId
+  setTimeout(() => {
+    void worker(runId)
   }, 0)
 }
 

frontend/src/repositories/Document.ts

@@ -59,6 +59,9 @@ export class Doc {
     if (this.dir) return false
     return ['jpg', 'jpeg', 'png', 'gif', 'webp', 'avif', 'heic', 'heif', 'svg'].includes(this.ext)
   }
+  get complete(): boolean {
+    return !this.ghost && (this.dir || this.size <= this.allocated)
+  }
   get previewable(): boolean {
     // Folders cannot be previewable
     if (this.dir) return false
@@ -67,6 +70,7 @@ export class Doc {
     return ['mp4', 'mkv', 'webm', 'ogg', 'mp3', 'flac', 'aac', 'pdf'].includes(this.ext)
   }
   get previewurl(): string {
+    if (!this.complete || !this.previewable) return ''
     return this.url.replace(/^\/files/, '/preview')
   }
   get ext(): string {

frontend/src/repositories/WS.ts

@@ -3,7 +3,6 @@ import { showAuthIframe, AuthCancelledError, isAuthIframeOpen } from 'paskia'
 import type { FileEntry, UpdateEntry, errorEvent } from "./Document"
 
 export const controlUrl = '/api/control'
-export const uploadUrl = '/api/upload'
 export const watchUrl = '/api/watch'
 
 let tree = [] as FileEntry[]

frontend/src/stores/main.ts

@@ -378,22 +378,16 @@ export const useMainStore = defineStore('main', {
       // What did we not select?
       for (const key of selected) if (!found.has(key)) ret.missing.add(key)
       // Build a flat list including contents recursively
-      const relnames = new Set<string>()
-      function add(rel: string, full: string, doc: Doc) {
-        if (!doc.dir && relnames.has(rel)) throw Error(`Multiple selections conflict for: ${rel}`)
-        relnames.add(rel)
-        ret.recursive.push([rel, full, doc])
-      }
       for (const key of ret.keys) {
         const base = ret.docs[key]!
         const basepath = base.loc ? `${base.loc}/${base.name}` : base.name
         const nremove = base.loc.length
-        add(base.name, basepath, base)
+        ret.recursive.push([base.name, basepath, base])
         for (const doc of docs) {
           if (doc.loc === basepath || doc.loc.startsWith(basepath) && doc.loc[basepath.length] === '/') {
             const full = doc.loc ? `${doc.loc}/${doc.name}` : doc.name
             const rel = full.slice(nremove)
-            add(rel, full, doc)
+            ret.recursive.push([rel, full, doc])
           }
         }
       }

pyproject.toml

@@ -41,10 +41,12 @@ dependencies = [
     "pillow-heif>=1.1.0",
     "pyjwt>=2.10.1",
     "pymupdf>=1.26.3",
+    "pyvips[binary]>=3.1.1",
     "sanic>=25.12.0",
     "setproctitle>=1.3.6",
     "stream-zip>=0.0.83",
     "tomli_w>=1.2.0",
+    "tracerite>=2.3.1",
     "zstandard>=0.24.0",
 ]
 

scripts/preview_images.py

@@ -0,0 +1,102 @@
+import argparse
+import mimetypes
+from pathlib import Path
+
+from cista.preview import process_image_with_timing
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Generate image previews for all files in a folder, one at a time.",
+    )
+    parser.add_argument("folder", type=Path, help="Folder to scan recursively")
+    parser.add_argument(
+        "--px",
+        type=int,
+        default=1024,
+        help="Maximum preview dimension in pixels (default: 1024)",
+    )
+    parser.add_argument(
+        "--quality",
+        type=int,
+        default=60,
+        help="AVIF quality passed to preview generation (default: 60)",
+    )
+    return parser.parse_args()
+
+
+def is_image_file(path: Path) -> bool:
+    mime_type, _ = mimetypes.guess_type(path.name)
+    return bool(mime_type and mime_type.startswith("image/"))
+
+
+def main() -> int:
+    args = parse_args()
+    folder = args.folder.resolve()
+    if not folder.is_dir():
+        raise SystemExit(f"Not a directory: {folder}")
+
+    files = sorted(
+        path for path in folder.rglob("*") if path.is_file() and is_image_file(path)
+    )
+    if not files:
+        print(f"No image files found under {folder}")
+        return 0
+
+    total_files = 0
+    total_bytes = 0
+    total_timing_slots: list[float] = []
+    total_preview_ms: float = 0.0
+    failures = 0
+
+    print(f"Scanning {folder}")
+    print(f"Generating previews for {len(files)} image files")
+
+    for path in files:
+        total_files += 1
+        rel = path.relative_to(folder)
+        try:
+            preview, timing = process_image_with_timing(
+                path,
+                maxsize=args.px,
+                quality=args.quality,
+            )
+        except Exception as exc:
+            failures += 1
+            print(f"FAIL  {rel}  error={exc}")
+            continue
+
+        total_bytes += len(preview)
+        timings = timing.timings or []
+        if len(total_timing_slots) < len(timings):
+            total_timing_slots.extend([0.0] * (len(timings) - len(total_timing_slots)))
+        for i, value in enumerate(timings):
+            total_timing_slots[i] += value
+        total_ms = sum(timings)
+        total_preview_ms += total_ms
+
+        detail = " / ".join(f"{value:.1f}ms" for value in timings)
+        if detail:
+            detail = f"timings={detail}  total={total_ms:.1f}ms"
+        else:
+            detail = f"total={total_ms:.1f}ms"
+        print(f"OK    {rel}  backend={timing.backend}  bytes={len(preview)}  {detail}")
+
+    completed = total_files - failures
+    print()
+    print("Summary")
+    print(f"  files={total_files}")
+    print(f"  completed={completed}")
+    print(f"  failed={failures}")
+    print(f"  preview_bytes={total_bytes}")
+    if completed:
+        if total_timing_slots:
+            for i, value in enumerate(total_timing_slots, start=1):
+                print(f"  timing{i}_total_ms={value:.1f}")
+        print(f"  preview_total_ms={total_preview_ms:.1f}")
+        print(f"  preview_avg_ms={total_preview_ms / completed:.1f}")
+    return 0 if failures == 0 else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())