pagerite/pagerite/migrations.py

"""Kanta schema migrations, discovered by name (``migrate_vN``).

Each function receives the raw state dict (JSON-level: bytes are base64
strings, datetimes RFC 3339 strings, struct fields with default values
omitted) before it is decoded into ``Data`` structs, and runs exactly once
per database based on its recorded version.

All storage/schema upgrades live here — including on-disk file work, which
runs through app.py's file store (imported lazily: app.py owns the store
and passes this module to Kanta; at migration time, during lifespan
``kanta.open()``, the app module is fully loaded).
"""

import base64
import re
from pathlib import Path

from pagerite.data import prettify


def _append_order(nodes: dict) -> float:
    """Raw-dict equivalent of data.append_order (order keys may be absent)."""
    return max((n.get("order", 0) for n in nodes.values()), default=0) + 1


def _ensure(menu: dict, path: str) -> dict:
    """Raw-dict equivalent of app._ensure: the node dict at ``path``,
    creating it and any missing ancestors (content-less category labels)
    appended at the end of their level."""
    nodes = menu
    node = None
    for seg in path.split("/"):
        node = nodes.get(seg)
        if node is None:
            node = {"title": prettify(seg), "order": _append_order(nodes)}
            nodes[seg] = node
        nodes = node.setdefault("children", {})
    return node


def migrate_v1(d: dict) -> None:
    """Move in-database file blobs to the on-disk content-addressed store,
    and rebuild the legacy flat page store (``pages``) as the menu tree."""
    files = d.pop("files", None)
    if files:
        from pagerite.app import file_store

        for name, body in files.items():
            if isinstance(body, str):  # JSON-level bytes are base64 strings
                body = base64.b64decode(body)
            file_store.put(name, body)
    pages = d.pop("pages", None)
    if not pages:
        return
    menu = d.setdefault("menu", {})
    for path, page in pages.items():
        node = _ensure(menu, path)
        node["title"] = page["title"]
        node["content"] = page["markdown"]
        for key in ("banner", "published", "order", "created", "modified"):
            if key in page:
                node[key] = page[key]


#: Extension-less file links: uploaded images are linked as /_f/<hash>
#: and the server negotiates avif/webp/jpg from the Accept header.
_DERIVATIVE_LINK = re.compile(r"(/_f/[0-9a-f]{12})\.(?:avif|webp)\b")


def _backfill_derivatives() -> None:
    """Create missing AVIF/WebP/JPEG derivatives for files stored before
    they were introduced (older uploads may have only the original plus
    AVIF, and SVGs no raster variants at all).  WebP/JPEG are re-encoded
    from an existing AVIF when available, everything else from the
    original (SVGs rasterized first)."""
    from pagerite import app

    file_store = app.file_store
    try:
        paths = [f for f in file_store.path.iterdir() if f.is_file()]
    except FileNotFoundError:
        return
    groups: dict[str, list[Path]] = {}
    for p in paths:
        groups.setdefault(p.name.partition(".")[0], []).append(p)
    for digest, files in groups.items():
        names = {p.name for p in files}
        source = next(
            (p for p in files if ".orig." in p.name or p.suffix == ".svg"), None
        )
        if source is None:
            continue  # plain as-is file, no derivatives to make
        avif = file_store.get(f"{digest}.avif")
        if avif is None:
            ext = source.suffix
            body = source.read_bytes()
            if ext == ".svg":
                png = app._svg_to_png(body, app.IMAGE_MAXSIZE)
                if png is None:
                    continue
                body, ext = png, ".png"
            converted = app._to_avif(body, ext)
            if converted is None:
                continue
            file_store.put(f"{digest}.avif", converted)
            avif = file_store.get(f"{digest}.avif")
        for fmt, quality in (
            ("webp", app.IMAGE_WEBP_QUALITY),
            ("jpg", app.IMAGE_JPG_QUALITY),
        ):
            if f"{digest}.{fmt}" not in names:
                file_store.put(
                    f"{digest}.{fmt}", app._avif_to_format(avif[0], f".{fmt}", quality)
                )


def migrate_v2(d: dict) -> None:
    """Extension-less image links: strip .avif/.webp extensions from /_f/
    links in page content and banners (the server now negotiates the format
    by Accept header), backfill missing AVIF/WebP/JPEG derivatives on disk,
    and drop the obsolete render-counter field ``version`` (invalidation is
    an in-memory concern now, not database state)."""

    def walk(nodes: dict) -> None:
        for node in nodes.values():
            for field in ("content", "banner"):
                if isinstance(node.get(field), str):
                    node[field] = _DERIVATIVE_LINK.sub(r"\1", node[field])
            walk(node.get("children") or {})

    walk(d.get("menu") or {})
    d.pop("version", None)
    _backfill_derivatives()