Preview: add pdf+pyvips encode path and simplify timings

- MediaPreview.vue: only render <img> for non-video, non-audio items
  (video/audio already have their own player containers)
- Document.ts: limit previewurl to previewable files (was checking only .complete)

README.md

@@ -20,6 +20,11 @@ Experience Cista by visiting [Cista Demo](https://drop.zi.fi) for a test run and
 
 We recommend using [UV](https://docs.astral.sh/uv/getting-started/installation/) to directly run Cista:
 
+Try it out locally at http://localhost:8000 (serves the current directory):
+```fish
+uvx cista
+```
+
 Create an account: (otherwise the server is public for all)
 ```fish
 uvx cista --user yourname --privileged

cista/__main__.py

@@ -61,10 +61,12 @@ doc = """\
 Usage:
   cista [-c <confdir>] [-l <host>] [--import-droppy] [--dev] [<path>]
   cista [-c <confdir>] --user <name> [--privileged] [--password]
+  cista --version
 
 Options:
   -c CONFDIR        Custom config directory
-  -l LISTEN-ADDR    Listen on
+  -l, --listen LISTEN-ADDR
+                    Listen on
                        :8989 (localhost port, plain http)
                        <addr>:3000 (bind another address, port)
                        /path/to/unix.sock (unix socket)
@@ -117,7 +119,7 @@ def _main():
     args = docopt(doc)
     if args["--user"]:
         return _user(args)
-    listen = args["-l"]
+    listen = args["--listen"]
     # Validate arguments first
     if args["<path>"]:
         path = Path(args["<path>"]).resolve()

cista/app.py

@@ -1,6 +1,7 @@
 import asyncio
 import datetime
 import mimetypes
+import time
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import cpu_count
 from pathlib import Path, PurePath, PurePosixPath
@@ -18,13 +19,19 @@ from stream_zip import ZIP_AUTO, stream_zip
 from zstandard import ZstdCompressor
 
 from cista import auth, config, preview, session, sso, watching
+from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.api import bp
+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 +49,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 +62,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 +71,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 +89,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."""

cista/preview.py

@@ -2,25 +2,34 @@ import asyncio
 import gc
 import io
 import mimetypes
+import os
+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 blake3 import blake3
 from PIL import Image
 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()
@@ -70,6 +79,208 @@ 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
+FORCE_PIL = os.environ.get("CISTA_PIL") == "1"
+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")
+        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(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(filepath.name)
+        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,6 +288,24 @@ async def verify_preview(request):
     await auth.verify(request)
 
 
+class PreviewTimeout(Exception):
+    """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
+
+
+class PreviewError(Exception):
+    """Raised when the preview subprocess exits with a non-zero status."""
+
+
+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(filepath.name)
+    return await _preview_pool.run(filepath, quality, maxsize, maxzoom)
+
+
 # Map EXIF Orientation value to a corresponding PIL transpose
 EXIF_ORI = {
     2: Image.Transpose.FLIP_LEFT_RIGHT,
@@ -89,6 +318,19 @@ EXIF_ORI = {
 }
 
 
+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>")
 async def preview(req, path):
     """Preview a file"""
@@ -97,7 +339,14 @@ async def preview(req, path):
     quality = int(req.args.get("q", 60))
     rel = PurePosixPath(sanitize(unquote(path)))
     filepath = config.config.path / rel
-    stat = filepath.lstat()
+    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 +361,40 @@ 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(
-        req.app.ctx.threadexec, dispatch, filepath, quality, maxsize, maxzoom
-    )
+    try:
+        img, preview_resp = await _run_preview_process(
+            filepath, quality, maxsize, maxzoom
+        )
+    except PreviewTimeout:
+        return empty(504)
+    except PreviewError:
+        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))
@@ -140,19 +404,54 @@ async def preview(req, path):
 
 def dispatch(path, quality, maxsize, maxzoom):
     try:
-        if path.suffix.lower() in (".pdf", ".xps", ".epub", ".mobi"):
+        if path.suffix.lower() in DOC_PREVIEW_SUFFIXES:
             return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
-        type, _ = mimetypes.guess_type(path.name)
-        if type and type.startswith("video/"):
+        mime_type, _ = mimetypes.guess_type(path.name)
+        if mime_type and mime_type.startswith("video/"):
             return process_video(path, quality=quality, maxsize=maxsize)
-        return process_image(path, quality=quality, maxsize=maxsize)
+        if mime_type and mime_type.startswith("image/"):
+            return process_image(path, quality=quality, maxsize=maxsize)
     except ValueError as e:
         logger.warning(f"Cannot generate preview for {path}: {e}")
     except Exception as e:
         logger.exception(f"Error generating preview for {path}: {e}")
+    return None, PreviewResponse(ok=False)
 
 
 def process_image(path, *, maxsize, quality):
+    return process_image_with_timing(path, maxsize=maxsize, quality=quality)
+
+
+def process_image_with_timing(path, *, maxsize, quality):
+    if FORCE_PIL:
+        return process_image_pillow(path, maxsize=maxsize, quality=quality)
+    return process_image_pyvips(path, maxsize=maxsize, quality=quality)
+
+
+def process_image_pyvips(path, *, maxsize, quality):
+    t_start = perf_counter()
+    img = pyvips.Image.new_from_file(str(path), access="sequential")
+    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_image_pillow(path, *, maxsize, quality):
     t_load = perf_counter()
     with Image.open(path) as img:
         # Force decode to include I/O in load timing
@@ -168,7 +467,14 @@ def process_image(path, *, maxsize, quality):
         # Save as AVIF
         imgdata = io.BytesIO()
         t_save = perf_counter()
-        img.save(imgdata, format="avif", quality=quality, speed=10, max_threads=1)
+        img.save(
+            imgdata,
+            format="avif",
+            quality=quality,
+            speed=10,
+            max_threads=1,
+            avif=1,
+        )
 
     t_end = perf_counter()
     ret = imgdata.getvalue()
@@ -176,16 +482,13 @@ def process_image(path, *, maxsize, quality):
     load_ms = (t_proc - t_load) * 1000
     proc_ms = (t_save - t_proc) * 1000
     save_ms = (t_end - t_save) * 1000
-    logger.debug(
-        "Preview image %s: load=%.1fms process=%.1fms save=%.1fms",
-        path.name,
-        load_ms,
-        proc_ms,
-        save_ms,
+    return ret, PreviewResponse(
+        ok=True,
+        mime="image/avif",
+        backend="pillow",
+        timings=[round(load_ms, 1), round(proc_ms, 1), round(save_ms, 1)],
     )
 
-    return ret
-
 
 def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
     t_load_start = perf_counter()
@@ -198,16 +501,30 @@ 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)
+    if FORCE_PIL:
+        ret = pix.pil_tobytes(
+            format="avif", quality=quality, speed=10, max_threads=1, avif=1
+        )
+        backend = "pdf"
+    else:
+        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(
-        "Preview pdf %s: load+render=%.1fms save=%.1fms",
-        path.name,
-        (t_load_end - t_load_start) * 1000,
-        (t_save_end - t_save_start) * 1000,
+    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),
+        ],
     )
-    return ret
 
 
 def process_video(path, *, maxsize, quality):
@@ -220,7 +537,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 +635,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(
-        "Preview video %s: load+decode=%.1fms save=%.1fms",
-        path.name,
-        (t_load_end - t_load_start) * 1000,
-        (t_save_end - t_save_start) * 1000,
+    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),
+        ],
     )
     del imgdata, istream, ostream, icc, occ, frame
     gc.collect()
-    return ret
+    return ret, resp

cista/preview_worker.py

@@ -0,0 +1,93 @@
+"""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 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
+
+
+_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
+        try:
+            req = _dec_req.decode(line)
+            result, resp = dispatch(
+                Path(req.path), req.quality, req.maxsize, req.maxzoom
+            )
+            _write_response(resp, result or b"")
+        except Exception as e:
+            _write_response(PreviewResponse(ok=False, error=str(e)), b"")
+
+
+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/sanic_logging.py

@@ -0,0 +1,256 @@
+"""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, "▪️")
+        return f"{emoji}  {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()
@@ -62,11 +62,6 @@ def parse_listen(listen):
         return "http://localhost", {"unix": unix.as_posix()}
 
     host, port = ep["host"], ep["port"]
-    # When binding all interfaces, use single_listener=False for Sanic
     if len(endpoints) > 1:
-        return f"http://localhost:{port}", {
-            "host": host,
-            "port": port,
-            "single_listener": False,
-        }
+        return f"http://localhost:{port}", {"host": host, "port": port}
     return f"http://{host}:{port}", {"host": host, "port": port}

cista/sso.py

@@ -15,7 +15,8 @@ import re
 
 import httpx
 import websockets
-from sanic import Blueprint
+from sanic import Blueprint, json
+from sanic import raw as raw_response
 from sanic.exceptions import Forbidden, SanicException, Unauthorized
 from sanic.log import logger
 
@@ -220,8 +221,6 @@ async def proxy_auth_request(request):
                 if key.lower() not in resp_hop_by_hop
             ]
 
-            from sanic import raw as raw_response
-
             return raw_response(
                 raw_content,
                 status=response.status_code,
@@ -231,35 +230,31 @@ async def proxy_auth_request(request):
 
     except httpx.RequestError as e:
         logger.error(f"Auth proxy request failed: {e}")
-        from sanic import json
 
         return json(
-            {"detail": "Authentication service unavailable", "error": str(e)},
+            {"detail": "Authentication service unavailable"},
             status=503,
         )
 
 
 async def proxy_auth_websocket(request, ws):
     """Proxy a WebSocket connection to the auth backend."""
-    path = request.path
-    query_string = request.query_string
-    ws_backend = PASKIA_BACKEND_URL.replace("http://", "ws://").replace(
-        "https://", "wss://"
-    )
-    url = f"{ws_backend}{path}"
-    if query_string:
-        url = f"{url}?{query_string}"
+    url = f"ws{PASKIA_BACKEND_URL.removeprefix('http')}{request.path}"
+    if request.query_string:
+        url = f"{url}?{request.query_string}"
 
     additional_headers = {}
     if "cookie" in request.headers:
         additional_headers["cookie"] = request.headers["cookie"]
     if "authorization" in request.headers:
         additional_headers["authorization"] = request.headers["authorization"]
+    if "host" in request.headers:
+        additional_headers["host"] = request.headers["host"]
     if "origin" in request.headers:
         additional_headers["origin"] = request.headers["origin"]
     if "user-agent" in request.headers:
         additional_headers["user-agent"] = request.headers["user-agent"]
-    additional_headers["x-forwarded-for"] = request.ip
+    additional_headers["x-forwarded-for"] = request.client_ip.strip("[]")
     additional_headers["x-forwarded-host"] = request.host
     additional_headers["x-forwarded-proto"] = request.scheme
 
@@ -291,23 +286,20 @@ async def proxy_auth_websocket(request, ws):
         logger.error(f"WebSocket proxy to {url} failed: {e}")
 
 
-def _is_websocket_request(request) -> bool:
-    """Check if the request is a WebSocket upgrade request."""
-    connection = request.headers.get("connection", "").lower()
-    upgrade = request.headers.get("upgrade", "").lower()
-    connection_tokens = [t.strip() for t in connection.split(",")]
-    return "upgrade" in connection_tokens and upgrade == "websocket"
+# Blueprint for auth proxy routes (only registered when paskia_enabled())
+bp = Blueprint("sso", url_prefix="/auth")
 
 
-async def _handle_websocket_upgrade(request):
-    """Handle WebSocket upgrade and proxy the connection."""
-    protocol = request.transport.get_protocol()
-    ws = await protocol.websocket_handshake(request, subprotocols=None)
+@bp.websocket("/ws/<path:path>")
+async def auth_websocket_proxy(request, ws, path=""):
+    """Proxy WebSocket connections to the auth backend."""
     await proxy_auth_websocket(request, ws)
 
 
-# Blueprint for auth proxy routes (only registered when paskia_enabled())
-bp = Blueprint("sso", url_prefix="/auth")
+@bp.websocket("/ws/")
+async def auth_websocket_proxy_root(request, ws):
+    """Proxy root WebSocket connections to the auth backend."""
+    await proxy_auth_websocket(request, ws)
 
 
 @bp.route(
@@ -315,20 +307,10 @@ bp = Blueprint("sso", url_prefix="/auth")
 )
 async def auth_proxy(request, path=""):
     """Proxy all auth requests to the auth backend."""
-    if _is_websocket_request(request):
-        await _handle_websocket_upgrade(request)
-        from sanic import empty
-
-        return empty()
     return await proxy_auth_request(request)
 
 
 @bp.route("/", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"])
 async def auth_proxy_root(request):
     """Proxy root auth requests to the auth backend."""
-    if _is_websocket_request(request):
-        await _handle_websocket_upgrade(request)
-        from sanic import empty
-
-        return empty()
     return await proxy_auth_request(request)

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

cista/watching.py

@@ -17,6 +17,33 @@ from cista import config
 from cista.fileio import fuid
 from cista.protocol import FileEntry, Space, UpdDel, UpdIns, UpdKeep
 
+# Platform-specific allocated size calculation
+if sys.platform == "win32":
+    import ctypes
+    from ctypes import wintypes
+
+    kernel32 = ctypes.windll.kernel32
+    GetCompressedFileSizeW = kernel32.GetCompressedFileSizeW
+    GetCompressedFileSizeW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(wintypes.DWORD)]
+    GetCompressedFileSizeW.restype = wintypes.DWORD
+    INVALID_FILE_SIZE = 0xFFFFFFFF
+
+    def get_allocated_size(path: Path, st: stat_result) -> int:
+        """Get actual disk allocation on Windows using GetCompressedFileSizeW."""
+        high = wintypes.DWORD()
+        low = GetCompressedFileSizeW(str(path), ctypes.byref(high))
+        if low == INVALID_FILE_SIZE and ctypes.get_last_error() != 0:
+            raise OSError(f"GetCompressedFileSizeW failed for {path}")
+        return (high.value << 32) + low
+
+else:
+
+    def get_allocated_size(path: Path, st: stat_result) -> int:
+        """Get actual disk allocation on Unix using st_blocks."""
+        # st_blocks is in 512-byte units
+        return st.st_blocks * 512
+
+
 pubsub = {}
 sortkey = natsort_keygen(alg=ns.LOCALE)
 
@@ -148,8 +175,11 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
     try:
         st = stat or path.stat()
         isfile = int(not S_ISDIR(st.st_mode))
-        # st_blocks is in 512-byte units
-        allocated = st.st_blocks * 512 if isfile else 0
+        try:
+            allocated = get_allocated_size(path, st) if isfile else 0
+        except Exception:
+            logger.exception(f"get_allocated_size failed for {path}")
+            allocated = st.st_size if isfile else 0
         entry = FileEntry(
             level=len(rel.parts),
             name=rel.name,

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">
-    <video ref=vid :src=doc.url :poster=poster preload=none @play=onplay @pause=onpaused @ended=next @seeking=media!.play()></video>
+  <div v-else-if=video() class="video-container" :class="{ pending: !doc.complete }">
+    <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

@@ -131,6 +131,7 @@ const uploadCloudFiles = (files: CloudFile[]) => {
 
 const cancelUploads = () => {
   upqueue = []
+  blockQueue = []
   statReset()
 }
 
@@ -162,16 +163,42 @@ setInterval(() => {
     store.uprogress.statdur *= .9
   }
 }, 100)
+// Track uploaded bytes for each file to handle out-of-order uploads
+const uploadedBytes = new Map<string, Set<number>>()
+
 const statUpdate = ({name, size, start, end}: {name: string, size: number, start: number, end: number}) => {
   if (name !== store.uprogress.filename) return  // If stats have been reset
   const now = Date.now()
-  store.uprogress.xfer = store.uprogress.filestart + end
-  store.uprogress.filepos = end
+
+  // Track which bytes have been uploaded (using start to end range)
+  if (!uploadedBytes.has(name)) uploadedBytes.set(name, new Set())
+  const uploaded = uploadedBytes.get(name)!
+  const blockSize = 1 << 20
+
+  // Mark all bytes in this block as uploaded
+  for (let i = start; i < end; i += blockSize) {
+    uploaded.add(i)
+  }
+
+  // Calculate total uploaded bytes for progress
+  let totalUploaded = 0
+  for (let i = 0; i < size; i += blockSize) {
+    if (uploaded.has(i)) totalUploaded += blockSize
+  }
+
+  store.uprogress.xfer = store.uprogress.filestart + totalUploaded
+  store.uprogress.filepos = totalUploaded
   store.uprogress.statbytes += end - start
   store.uprogress.statdur += now - store.uprogress.tlast
   store.uprogress.tlast = now
-  // File finished?
-  if (end === size) {
+
+  // Check if file is fully uploaded by examining the block queue
+  const currentUpload = blockQueue[0]
+  if (!currentUpload) return
+
+  if (currentUpload.file.cloudName === name && currentUpload.blockIndex >= 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()
@@ -198,6 +225,42 @@ const statsAdd = (f: CloudFile[]) => {
 }
 let upqueue = [] as CloudFile[]
 
+// Helper function to get upload blocks for a file, prioritizing final 4 blocks if file >= 32 MiB
+const getUploadBlocks = (file: CloudFile): {start: number, end: number}[] => {
+  const BLOCK_SIZE = 1 << 20  // 1 MiB
+  const MIN_SIZE_FOR_REORDER = 32 * BLOCK_SIZE  // 32 MiB = 33554432 bytes
+  const FINAL_BLOCKS_COUNT = 2
+
+  const fileSize = file.file.size
+  const blocks: {start: number, end: number}[] = []
+
+  if (fileSize >= MIN_SIZE_FOR_REORDER) {
+    // File is large enough, prioritize final blocks
+    const finalBlocksStart = fileSize - (FINAL_BLOCKS_COUNT * BLOCK_SIZE)
+
+    // 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})
+    }
+
+    // Add remaining blocks from beginning
+    for (let start = 0; start < finalBlocksStart; start += BLOCK_SIZE) {
+      const end = Math.min(start + BLOCK_SIZE, finalBlocksStart)
+      blocks.push({start, end})
+    }
+  } else {
+    // File is smaller, use sequential upload
+    for (let start = 0; start < fileSize; start += BLOCK_SIZE) {
+      const end = Math.min(start + BLOCK_SIZE, fileSize)
+      blocks.push({start, end})
+    }
+  }
+
+  return blocks
+}
+
 // TODO: Rewrite as WebSocket class
 const WSCreate = async () => await new Promise<WebSocket>(resolve => {
   const ws = connect(uploadUrl, {
@@ -235,31 +298,58 @@ const WSCreate = async () => await new Promise<WebSocket>(resolve => {
     ws.send(data)
   }
 })
+
+type BlockUpload = {
+  file: CloudFile
+  blocks: {start: number, end: number}[]
+  blockIndex: number
+}
+
+let blockQueue = [] as BlockUpload[]
+
 const worker = async () => {
   const ws = await WSCreate()
-  while (upqueue.length) {
-    const f = upqueue[0]!
-    const start = f.cloudPos
-    const end = Math.min(f.file.size, start + (1<<20))
-    const control = { name: f.cloudName, size: f.file.size, start, end }
-    const data = f.file.slice(start, end)
-    f.cloudPos = end
+  while (blockQueue.length) {
+    const upload = blockQueue[0]!
+    const f = upload.file
+    const block = upload.blocks[upload.blockIndex]!
+
+    const control = { name: f.cloudName, size: f.file.size, start: block.start, end: block.end }
+    const data = f.file.slice(block.start, block.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()
+
+    // Move to next block
+    upload.blockIndex++
+    if (upload.blockIndex >= upload.blocks.length) {
+      // File upload complete
+      blockQueue.shift()
+    }
   }
-  if (upqueue.length) startWorker()
+  if (blockQueue.length) startWorker()
   store.uprogress.status = "idle"
   workerRunning = false
 }
 let workerRunning: any = false
 const startWorker = () => {
   if (workerRunning === false) workerRunning = setTimeout(() => {
-    workerRunning = true
-    worker()
+    // Convert new CloudFile entries to BlockUpload entries
+    while (upqueue.length) {
+      const file = upqueue.shift()!
+      const blocks = getUploadBlocks(file)
+      blockQueue.push({ file, blocks, blockIndex: 0 })
+    }
+
+    if (blockQueue.length) {
+      workerRunning = true
+      worker()
+    } else {
+      workerRunning = false
+    }
   }, 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/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",
 ]
 
@@ -114,6 +116,7 @@ filterwarnings = [
 ]
 
 [tool.ruff.lint]
+extend-select = ["E402"]
 isort.known-first-party = ["cista"]
 per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004"]
 per-file-ignores."scripts/*" = ["T20"]
@@ -121,16 +124,13 @@ per-file-ignores."scripts/*" = ["T20"]
 [dependency-groups]
 dev = [
     "pytest>=8.4.1",
+    "pytest-asyncio>=0.25.0",
+    "pytest-cov>=7.0.0",
     "ruff>=0.8.0",
     "mypy>=1.13.0",
     "pre-commit>=4.0.0",
     "httpx>=0.28.1",
 ]
-test = [
-    "pytest>=8.4.1",
-    "pytest-cov>=6.0.0",
-    "pytest-asyncio>=0.25.0",
-]
 
 [tool.coverage.run]
 source = ["cista"]

scripts/fastapi-vue/buildutil.py

@@ -134,7 +134,7 @@ def find_dev_tool() -> list[str]:
     Raises RuntimeError if no runtime is found.
     """
     dev_args = {
-        "deno": ("run", "dev", "--"),
+        "deno": ("run", "-A", "npm:vite"),
         "npm": ("--silent", "run", "dev", "--"),
         "bun": ("run", "dev", "--"),
     }

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())