arthur/clips/views.py

"""The API's implementation.

Two things in here are load-bearing and neither is Django.

THE SERVER VERIFIES EVERY TIER-2 KEY IT IS HANDED. A key is the sha256 of a
canonical descriptor, and this recomputes it and refuses a mismatch. That is what
makes content addressing a property of the system rather than a convention in the
client: nothing can store bytes under a name that does not describe them.

It hashes THE TEXT IT WAS SENT rather than re-rendering the descriptor from parsed
values, and that is the honest arrangement rather than a shortcut. JS prints an
integral double as `1` and Python prints `1.0`, so a scheme where both sides
re-render the numbers would disagree on the first parameter whose value happens to
be whole — and the failure would be an upload that 409s with nothing wrong. The
bytes are the contract; the schema on top of them is a convention, and the two
fields this file actually reads out of that schema are checked separately.

AND IT REFUSES A BLOCK WHOSE ANALYSIS IT DOES NOT KNOW. Every block descriptor
names an analysis, and every analysis declares a detector and a VERSION. So the
chain from a stored block to the model version that produced it cannot be broken
by a client that forgot a step — which is the whole point of
docs/architecture.md's insistence that the cache key include the detector version.
A model upgrade that silently reused old landmarks would otherwise present as "the
tool got worse", with no event to attach it to.
"""
import hashlib
import json
import re
import zlib
from functools import lru_cache
from pathlib import Path
from uuid import UUID

from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import transaction
from django.http import FileResponse, HttpResponse, JsonResponse
from django.shortcuts import render
from django.views.decorators.http import require_http_methods

from . import blobs, extraction
from .models import Analysis, Block, Blob, Clip, Extraction, Footage, Leaf, Project, Revision, Source

KEY_LENGTH = 71  # "sha256:" + 64 hex


# ---------------------------------------------------------------------------
# helpers


def _body(request):
    try:
        return json.loads(request.body or b"{}")
    except json.JSONDecodeError as exc:
        raise Bad(f"the request body is not JSON: {exc}") from exc


class Bad(Exception):
    """A 400 with a message, raised where the problem is noticed."""

    def __init__(self, message, status=400, **detail):
        super().__init__(message)
        self.message = message
        self.status = status
        self.detail = detail


def _error(exc: Bad):
    return JsonResponse({"error": exc.message, **exc.detail}, status=exc.status)


def _check_key(key, descriptor):
    """The verification. A key is the sha256 of the descriptor stored beside it."""
    if not isinstance(key, str) or len(key) != KEY_LENGTH or not key.startswith("sha256:"):
        raise Bad(f"not a content address: {key!r}")
    if not isinstance(descriptor, str) or not descriptor:
        raise Bad("a key without its descriptor addresses nothing")
    actual = hashlib.sha256(descriptor.encode("utf-8")).hexdigest()
    if actual != key[7:]:
        raise Bad(
            "the key is not the hash of its descriptor",
            status=409,
            expected=f"sha256:{actual}",
            given=key,
        )
    try:
        return json.loads(descriptor)
    except json.JSONDecodeError as exc:
        raise Bad(f"the descriptor is not canonical JSON: {exc}") from exc


def _blob(b64, media_type="application/octet-stream"):
    import base64

    digest, size = blobs.write(base64.b64decode(b64))
    blob, _ = Blob.objects.get_or_create(
        digest=digest, defaults={"size": size, "media_type": media_type}
    )
    return blob


def _uploaded_blob(upload, media_type="application/octet-stream"):
    digest, size = blobs.write_stream(upload.chunks())
    blob, _ = Blob.objects.get_or_create(
        digest=digest, defaults={"size": size, "media_type": media_type}
    )
    return blob


def _crop_blob(chunks):
    digest, size = blobs.write_compressed_stream(chunks)
    blob, _ = Blob.objects.get_or_create(
        digest=digest, defaults={"size": size, "media_type": blobs.CROP_MEDIA_TYPE}
    )
    return blob


# ---------------------------------------------------------------------------
# the page


def _asset_version(relative):
    """A static file's modification time, for its URL.

    The stylesheet and the bundle are served with `Last-Modified` and nothing
    else, so a browser is free to keep a stale copy on heuristic freshness — and
    new JavaScript over an old stylesheet renders a pane the stylesheet has never
    heard of as bare elements. A version in the URL makes each edit a new URL."""
    for root in settings.STATICFILES_DIRS:
        path = Path(root) / relative
        if path.exists():
            return str(int(path.stat().st_mtime))
    return "0"


def page(request):
    """The host page. This replaced `frontend/public/index.html` at step 9, and
    `:dev-http` in shadow-cljs.edn went away with it."""
    return render(request, "clips/index.html", {
        "css_version": _asset_version("arthur/app.css"),
        "js_version": _asset_version("arthur/js/main.js"),
    })


# ---------------------------------------------------------------------------
# the detector
#
# WHY THE SERVER ANSWERS THIS. The analysis key has to include the detector
# version, and a version string in the client is a string somebody has to remember
# to bump. The server serves the model, so it can hash the model — and then the
# version is a fact about the bytes that produced the landmarks rather than a
# claim about them.


@lru_cache(maxsize=4)
def _model_digest(path: str, mtime: float) -> str:
    return blobs.digest_file(Path(path))


def _package_version() -> str:
    pkg = Path(settings.BASE_DIR) / "frontend" / "package.json"
    try:
        deps = json.loads(pkg.read_text())["dependencies"]
        return deps["@mediapipe/tasks-vision"].lstrip("^~")
    except Exception:
        return "unknown"


@require_http_methods(["GET"])
def detector(request):
    model = Path(settings.BASE_DIR) / "frontend" / "public" / "mediapipe" / "face_landmarker.task"
    if not model.exists():
        # Honest rather than fatal: detection will fail at the MediaPipe boundary
        # with a better message than this one could give, and an analysis stamped
        # "unknown" is a take somebody can still look at and re-freeze later.
        return JsonResponse({"detector": "mediapipe", "version": "unknown", "model": None})
    digest = _model_digest(str(model), model.stat().st_mtime)
    return JsonResponse(
        {
            "detector": "mediapipe",
            # The package version AND the model's own hash. Either alone can change
            # while the other does not, and both change the landmarks.
            "version": f"{_package_version()}+{digest[:16]}",
            "model": f"sha256:{digest}",
        }
    )


# ---------------------------------------------------------------------------
# tier 3: footage
#
# THE MANIFEST NOW CARRIES URLS. It used to carry a directory and the loader built
# `frames/0001.png` itself, which quietly made the frame layout a shared secret
# between a shell script and a ClojureScript namespace. The server names every
# frame instead, so uploaded video and command-line bundles produce the same
# footage response without the client knowing where either stored its frames.


@require_http_methods(["GET", "POST"])
def sources(request):
    if request.method == "GET":
        return JsonResponse({"sources": [
            {"id": str(row.id), "filename": row.filename, "probe": row.probe}
            for row in Source.objects.order_by("-created")[:100]
        ]})
    upload = request.FILES.get("file")
    if upload is None:
        return JsonResponse({"error": "upload a video as the file field"}, status=400)
    try:
        digest, size = blobs.write_stream(upload.chunks())
        facts = extraction.probe(blobs.path_for(digest))
        blob, _ = Blob.objects.get_or_create(
            digest=digest, defaults={"size": size,
                                     "media_type": upload.content_type or "video/mp4"})
        row, created = Source.objects.get_or_create(
            blob=blob, defaults={"filename": Path(upload.name).name[:255], "probe": facts})
        return JsonResponse({"id": str(row.id), "digest": digest,
                             "filename": row.filename, "probe": row.probe,
                             "created": created}, status=201 if created else 200)
    except (ValueError, OSError) as exc:
        return JsonResponse({"error": str(exc)}, status=400)


def _extraction_json(row):
    return {"key": row.key, "source": str(row.source_id), "state": row.state,
            "progress": row.progress, "error": row.error,
            "footage": str(row.footage_id) if row.footage_id else None}


@require_http_methods(["POST"])
def extractions(request):
    try:
        data = _body(request)
        source_id = data.get("source")
        if not source_id:
            raise Bad("an extraction needs a source id")
        try:
            source = Source.objects.get(id=UUID(str(source_id)))
        except (ValueError, ValidationError, Source.DoesNotExist):
            raise Bad("no such source", status=404)
        settings = data.get("settings") or {}
        if settings != {}:
            raise Bad("extraction currently keeps the source frame rate; settings must be empty")
        key = extraction.extraction_key(source, settings)
        row, _ = Extraction.objects.get_or_create(
            key=key, defaults={"source": source, "settings": settings})
        if row.state != "done":
            extraction.enqueue(key)
        return JsonResponse(_extraction_json(row), status=202 if row.state != "done" else 200)
    except Bad as exc:
        return _error(exc)


@require_http_methods(["GET"])
def extraction_detail(request, key):
    try:
        return JsonResponse(_extraction_json(Extraction.objects.get(key=key)))
    except Extraction.DoesNotExist:
        return JsonResponse({"error": "no such extraction"}, status=404)


def _footage_json(footage: Footage, urls=True):
    out = {
        "id": str(footage.id),
        "label": footage.label or footage.source,
        "source": footage.source,
        "fps": footage.fps,
        "frames": footage.frames,
        "width": footage.width,
        "height": footage.height,
        "footage": f"sha256:{footage.digest}",
        "audio": f"/blob/{footage.audio.digest}",
        # The analysis source. `null` on footage ingested before the proxy
        # existed, which the loader reports as "re-extract this" rather than
        # failing somewhere inside MediaPipe.
        "video": f"/blob/{footage.video.digest}" if footage.video_id else None,
        # What the page actually decodes: one access unit per frame, no container.
        "stream": f"/blob/{footage.stream.digest}" if footage.stream_id else None,
        "feature-absence": footage.feature_absence or {},
    }
    if urls:
        out["urls"] = [f"/blob/{f.blob.digest}" for f in footage.frame_set.select_related("blob")]
    return out


@require_http_methods(["GET"])
def footage_list(request):
    return JsonResponse(
        {"footage": [_footage_json(f, urls=False) for f in Footage.objects.all()]}
    )


_SYMBOL_LEAF = re.compile(r"^clip/([^/]+)/symbol/([^/]+)$")


def _transit_fields(value, *keys):
    """Top-level fields of a transit map leaf, by keyword name. A leaf's own
    facts are a small flat map, so no key repeats and transit's cache never
    stands in for one; anything else reads as absent."""
    if not (isinstance(value, list) and value[:1] == ["^ "]):
        return {}
    pairs = dict(zip(value[1::2], value[2::2]))
    return {k: pairs.get(f"~:{k}") for k in keys}


@require_http_methods(["GET"])
def symbols(request):
    """Every symbol in every saved project, for the pool's all-assets folder.

    Read off the leaf PATHS rather than by loading documents: a symbol's own leaf
    is `clip/<cid>/symbol/<sid>`, so listing them is one query and no decoding
    beyond the name and length its value carries."""
    rows = []
    for leaf in Leaf.objects.filter(path__contains="/symbol/").select_related("project"):
        m = _SYMBOL_LEAF.match(leaf.path)
        if not m:
            continue
        fields = _transit_fields(leaf.value, "name", "frames")
        rows.append({
            "project": str(leaf.project_id),
            "project_name": leaf.project.name,
            "cid": m.group(1),
            "symbol": m.group(2),
            "name": fields.get("name") or m.group(2).replace("~", "/"),
            "frames": fields.get("frames"),
        })
    rows.sort(key=lambda r: (r["project_name"], r["project"], r["name"]))
    return JsonResponse({"symbols": rows})


@require_http_methods(["GET"])
def footage_detail(request, footage_id):
    try:
        footage = Footage.objects.select_related("audio", "video", "stream").get(id=footage_id)
    except Footage.DoesNotExist:
        return JsonResponse({"error": "no such footage"}, status=404)
    return JsonResponse(_footage_json(footage))


_RANGE = re.compile(r"^bytes=(\d*)-(\d*)$")


class _Slice:
    """A file, readable only up to `remaining` bytes from where it was seeked."""

    def __init__(self, handle, remaining):
        self.handle, self.remaining = handle, remaining

    def read(self, size=-1):
        if self.remaining <= 0:
            return b""
        if size < 0 or size > self.remaining:
            size = self.remaining
        data = self.handle.read(size)
        self.remaining -= len(data)
        return data

    def close(self):
        self.handle.close()


def _byte_range(header, size):
    """One `Range` header -> (start, end) inclusive, or None for the whole blob.

    A syntactically broken header is NOT an error: RFC 9110 says an unparsable
    Range is ignored and the whole representation is sent, which is what a client
    that meant nothing by it wants. `False` is the third answer — a range that
    parses and cannot be satisfied — because that one is a 416.
    """
    if not header:
        return None
    match = _RANGE.match(header.strip())
    if not match or match.group(1) == "" and match.group(2) == "":
        return None
    first, last = match.group(1), match.group(2)
    if first == "":
        # `bytes=-500`: the LAST 500 bytes, which is a different question.
        length = int(last)
        if length == 0:
            return False
        return (max(0, size - length), size - 1)
    start = int(first)
    end = int(last) if last else size - 1
    end = min(end, size - 1)
    if start >= size or start > end:
        return False
    return (start, end)


@require_http_methods(["GET"])
def blob(request, digest):
    """Raw bytes, immutable, and serveable a slice at a time.

    `immutable` is not optimism here, it is the definition: the name IS the hash of
    the content, so a cached copy cannot be stale. That is what makes serving a
    take's frames out of this cheap enough to do on every load.

    RANGE IS NOT AN OPTIMISATION HERE, IT IS THE FEATURE. Since the analysis source
    became a video file, a `<video>` element seeks this URL, and a media element
    that is handed 200OK with no `Accept-Ranges` cannot seek: it reports an empty
    `seekable` range, every `currentTime` write is a no-op, and detection then runs
    ninety times over frame one without anything raising. Django's `FileResponse`
    does not do this for us — there is no Range handling anywhere in it — so the
    absence of these thirty lines presents as "MediaPipe's video mode is broken".
    """
    try:
        row = Blob.objects.get(digest=digest)
        path = blobs.path_for(digest)
    except (Blob.DoesNotExist, ValueError):
        return JsonResponse({"error": "no such blob"}, status=404)

    size = path.stat().st_size
    span = _byte_range(request.headers.get("Range"), size)
    if span is False:
        response = HttpResponse(status=416)
        response["Content-Range"] = f"bytes */{size}"
    elif span is None:
        response = FileResponse(open(path, "rb"), content_type=row.media_type)
    else:
        start, end = span
        handle = open(path, "rb")
        handle.seek(start)
        response = FileResponse(_Slice(handle, end - start + 1),
                                status=206, content_type=row.media_type)
        response["Content-Range"] = f"bytes {start}-{end}/{size}"
        response["Content-Length"] = str(end - start + 1)
    response["Accept-Ranges"] = "bytes"
    response["Cache-Control"] = "public, max-age=31536000, immutable"
    response["ETag"] = f'"{digest}"'
    return response


# ---------------------------------------------------------------------------
# tier 2: analyses and blocks


@require_http_methods(["POST"])
def analyses(request):
    """Register an analysis artifact's identity. Idempotent: the same inputs are
    the same key are the same row."""
    try:
        data = _body(request)
        key = data.get("key")
        descriptor = data.get("descriptor")
        parsed = _check_key(key, descriptor)
        for field in ("detector", "version"):
            if not parsed.get(field):
                raise Bad(
                    f"the descriptor does not declare a {field}: a cache key that "
                    "omits the detector version lets a model upgrade silently reuse "
                    "old landmarks",
                    missing=field,
                )
        footage = None
        if data.get("footage"):
            digest = str(data["footage"]).removeprefix("sha256:")
            footage = Footage.objects.filter(digest=digest).first()
            if footage is None:
                raise Bad("the analysis names footage this server does not have",
                          footage=data["footage"])
        row, created = Analysis.objects.get_or_create(
            key=key,
            defaults={
                "descriptor": descriptor,
                "detector": parsed["detector"],
                "version": str(parsed["version"]),
                "footage": footage,
            },
        )
        return JsonResponse({"key": row.key, "created": created}, status=201 if created else 200)
    except Bad as exc:
        return _error(exc)


@require_http_methods(["GET", "PUT"])
def analysis_detail(request, key):
    try:
        row = Analysis.objects.get(key=key)
    except Analysis.DoesNotExist:
        return JsonResponse({"error": "no such analysis"}, status=404)
    if request.method == "GET":
        return JsonResponse({
            "key": row.key, "descriptor": row.descriptor,
            "detector": row.detector, "version": row.version,
            "footage": str(row.footage_id) if row.footage_id else None,
            "source_blocks": sorted(row.source_blocks.values_list("key", flat=True)),
        })
    try:
        keys = _body(request).get("source_blocks")
        roles = {"source/dense", "source/detected", "source/crops"}
        if (not isinstance(keys, list) or not keys
                or not all(isinstance(k, str) for k in keys) or len(set(keys)) != len(keys)):
            raise Bad("an analysis needs distinct source block keys")
        blocks = list(Block.objects.filter(key__in=keys))
        if len(blocks) != len(keys) or any(b.analysis_id != key for b in blocks):
            raise Bad("source blocks must exist and name this analysis")
        by_subject = {}
        for block in blocks:
            subjects = json.loads(block.descriptor).get("features", [])
            if (not isinstance(subjects, list) or len(subjects) > 1
                    or any(not isinstance(s, str) or not s for s in subjects)):
                raise Bad("a source block must name one subject")
            # Older single-face analyses used an empty feature list.
            by_subject.setdefault(tuple(subjects), []).append(block.role)
        if any(len(found) != len(roles) or set(found) != roles
               for found in by_subject.values()):
            raise Bad("each subject needs one block for each source role")
        with transaction.atomic():
            row = Analysis.objects.select_for_update().get(key=key)
            current = set(row.source_blocks.values_list("key", flat=True))
            if current and current != set(keys):
                raise Bad("the source blocks of an analysis are immutable", status=409)
            row.source_blocks.set(blocks)
        return JsonResponse({"key": key, "source_blocks": sorted(keys)})
    except Bad as exc:
        return _error(exc)


@require_http_methods(["POST"])
def blocks_missing(request):
    """Which of these keys the server does not have.

    The return on content addressing, as one request: a save uploads the blocks
    that are new and nothing else, so re-saving a document after a knob-free edit
    moves kilobytes.
    """
    try:
        keys = _body(request).get("keys") or []
        if not isinstance(keys, list):
            raise Bad("keys must be a list")
        have = set(Block.objects.filter(key__in=keys).values_list("key", flat=True))
        return JsonResponse({"missing": [k for k in keys if k not in have]})
    except Bad as exc:
        return _error(exc)


@require_http_methods(["POST"])
def blocks(request):
    """Store one dense block: its bytes, its optional absence mask, and the
    descriptor its key is the hash of."""
    try:
        multipart = request.content_type == "multipart/form-data"
        data = request.POST if multipart else _body(request)
        upload = request.FILES.get("data") if multipart else None
        state_upload = request.FILES.get("state") if multipart else None
        key = data.get("key")
        descriptor = data.get("descriptor")
        parsed = _check_key(key, descriptor)
        role = parsed.get("role")
        if not role:
            raise Bad("a block's descriptor names its role")
        if not parsed.get("layout", {}).get("type"):
            raise Bad(
                "a block's descriptor must say what its elements are: an Int16Array "
                "and a Float32Array over the same bytes are both valid readings and "
                "only one of them is the block"
            )
        analysis_key = parsed.get("analysis")
        analysis = Analysis.objects.filter(key=analysis_key).first()
        if analysis is None:
            raise Bad(
                "this block names an analysis the server does not know; register the "
                "analysis first, so that every stored block can name the detector "
                "version that produced it",
                analysis=analysis_key,
            )
        if not (upload and upload.size) and not data.get("data"):
            raise Bad("a block with no bytes")
        with transaction.atomic():
            if role == "source/crops":
                if upload:
                    data_blob = _crop_blob(upload.chunks())
                else:
                    import base64

                    data_blob = _crop_blob([base64.b64decode(data["data"])])
            else:
                data_blob = _uploaded_blob(upload) if upload else _blob(data["data"])
            row, created = Block.objects.get_or_create(
                key=key,
                defaults={
                    "descriptor": descriptor,
                    "role": role,
                    "analysis": analysis,
                    "data": data_blob,
                    "state": (_uploaded_blob(state_upload) if state_upload else
                              _blob(data["state"]) if data.get("state") else None),
                },
            )
        return JsonResponse({"key": row.key, "created": created}, status=201 if created else 200)
    except Bad as exc:
        return _error(exc)


@require_http_methods(["GET"])
def block_detail(request, key):
    import base64

    try:
        row = Block.objects.select_related("data", "state").get(key=key)
    except Block.DoesNotExist:
        return JsonResponse({"error": "no such block"}, status=404)
    data = blobs.read(row.data.digest)
    if row.data.media_type == blobs.CROP_MEDIA_TYPE:
        data = zlib.decompress(data)
    out = {
        "key": row.key,
        "descriptor": row.descriptor,
        "data": base64.b64encode(data).decode("ascii"),
    }
    if row.state_id:
        out["state"] = base64.b64encode(blobs.read(row.state.digest)).decode("ascii")
    response = JsonResponse(out)
    response["Cache-Control"] = "public, max-age=31536000, immutable"
    return response


# ---------------------------------------------------------------------------
# tier 1: projects, clips, leaves


def _project_json(project: Project):
    leaves = list(project.leaves.all())
    clips = []
    for clip in project.clips.all():
        prefix = f"clip/{clip.cid}/"
        clips.append(
            {
                "cid": clip.cid,
                "name": clip.name,
                "footage": str(clip.footage_id) if clip.footage_id else None,
                "analysis": clip.analysis_id,
                "blocks": sorted(clip.blocks.values_list("key", flat=True)),
                "leaves": {leaf.path: leaf.value for leaf in leaves if leaf.path.startswith(prefix)},
            }
        )
    return {
        "id": str(project.id),
        "name": project.name,
        "schema_version": project.schema_version,
        "seq": project.seq,
        "palette": project.palette,
        "clips": clips,
    }


@require_http_methods(["GET", "POST"])
def projects(request):
    if request.method == "GET":
        return JsonResponse(
            {
                "projects": [
                    {"id": str(p.id), "name": p.name,
                     "schema_version": p.schema_version, "seq": p.seq,
                     "updated": p.updated.isoformat()}
                    for p in Project.objects.all()[:100]
                ]
            }
        )
    try:
        data = _body(request)
        project = Project.objects.create(name=data.get("name") or "untitled")
        return JsonResponse(_project_json(project), status=201)
    except Bad as exc:
        return _error(exc)


@require_http_methods(["GET", "PUT"])
def project_detail(request, project_id):
    try:
        project = Project.objects.get(id=project_id)
    except Project.DoesNotExist:
        return JsonResponse({"error": "no such project"}, status=404)
    if request.method == "GET":
        return JsonResponse(_project_json(project))
    try:
        return _save(project, _body(request))
    except Bad as exc:
        return _error(exc)


@transaction.atomic
def _save(project: Project, data):
    """A whole-document save: one clip's leaves replace that clip's leaves.

    SCOPED BY CLIP, not by project. A payload that carries clip `a` does not
    disturb clip `b`'s leaves, because a save is not the only way the document
    changes — a single-leaf conditional write is — and a save that cleared
    everything it did not mention would be a save that undoes a collaborator.

    A leaf whose value is unchanged keeps its VERSION. That is what makes the
    entity tag mean something: a save of a document where one channel moved
    invalidates one leaf's etag, not all four hundred.
    """
    if data.get("name"):
        project.name = data["name"]
    if data.get("palette"):
        project.palette = data["palette"]

    written, removed, unchanged = [], [], []
    for spec in data.get("clips") or []:
        cid = spec.get("cid")
        if not cid:
            raise Bad("every clip in a save names its cid")
        leaves = spec.get("leaves") or {}
        prefix = f"clip/{cid}/"
        for path in leaves:
            if not path.startswith(prefix):
                raise Bad(
                    f"leaf {path!r} is not addressed to clip {cid!r}",
                    clip=cid, path=path,
                )

        keys = spec.get("blocks") or []
        have = set(Block.objects.filter(key__in=keys).values_list("key", flat=True))
        if missing := [k for k in keys if k not in have]:
            # Referential integrity across the tiers, enforced where it can be:
            # a document that names blocks the server does not hold would load
            # into a blank stage on any other machine.
            raise Bad(
                "this clip names tier-2 blocks the server does not have; upload them "
                "before saving the document that points at them",
                status=409, missing=missing,
            )

        analysis = Analysis.objects.filter(key=spec.get("analysis")).first()
        footage = None
        if spec.get("footage"):
            footage = Footage.objects.filter(id=spec["footage"]).first()
        clip, _ = Clip.objects.update_or_create(
            project=project,
            cid=cid,
            defaults={"name": spec.get("name") or "", "analysis": analysis, "footage": footage},
        )
        clip.blocks.set(Block.objects.filter(key__in=keys))

        existing = {leaf.path: leaf for leaf in project.leaves.filter(path__startswith=prefix)}
        for path, value in leaves.items():
            leaf = existing.get(path)
            if leaf is None:
                Leaf.objects.create(project=project, path=path, value=value)
                written.append(path)
            elif leaf.value != value:
                leaf.value = value
                leaf.version += 1
                leaf.save(update_fields=["value", "version", "updated"])
                written.append(path)
            else:
                unchanged.append(path)
        for path, leaf in existing.items():
            if path not in leaves:
                leaf.delete()
                removed.append(path)

    seq = project.seq + 1
    project.seq = seq
    project.save()
    return JsonResponse(
        {
            "id": str(project.id),
            "schema_version": project.schema_version,
            "seq": seq,
            "written": sorted(written),
            "removed": sorted(removed),
            "unchanged": len(unchanged),
        }
    )


@require_http_methods(["GET", "PUT"])
def leaf_detail(request, project_id, leaf_path):
    """One leaf, conditionally.

    `If-Match` and a 409 whose body carries the CURRENT value, so the client can
    offer keep-mine / take-theirs. A PUT that replaced unconditionally is the bug
    docs/architecture.md calls out in tl: the loser's work disappears silently, and
    for a painted cel that is the class of bug that ends trust in a tool.
    """
    try:
        project = Project.objects.get(id=project_id)
    except Project.DoesNotExist:
        return JsonResponse({"error": "no such project"}, status=404)

    leaf = project.leaves.filter(path=leaf_path).first()
    if request.method == "GET":
        if leaf is None:
            return JsonResponse({"error": "no such leaf"}, status=404)
        response = JsonResponse({"path": leaf.path, "value": leaf.value, "version": leaf.version})
        response["ETag"] = leaf.etag
        return response

    try:
        data = _body(request)
    except Bad as exc:
        return _error(exc)
    if "value" not in data:
        return _error(Bad("a leaf write carries a value"))

    match = request.headers.get("If-Match")
    if leaf is None:
        # ANY `If-Match` on a leaf that does not exist is a failed precondition,
        # `*` included: RFC 7232 gives `*` the meaning "the resource must already
        # exist", which is exactly the write a client makes when it believes it is
        # editing something. Creating it instead would turn "somebody deleted this
        # node" into a silent resurrection.
        if match:
            return JsonResponse(
                {"error": "no such leaf", "path": leaf_path}, status=409
            )
        leaf = Leaf.objects.create(project=project, path=leaf_path, value=data["value"])
    else:
        if match and match not in ("*", leaf.etag):
            response = JsonResponse(
                {
                    "error": "stale write",
                    "path": leaf.path,
                    "version": leaf.version,
                    "value": leaf.value,
                },
                status=409,
            )
            response["ETag"] = leaf.etag
            return response
        leaf.value = data["value"]
        leaf.version += 1
        leaf.save(update_fields=["value", "version", "updated"])

    seq = project.bump()
    response = JsonResponse({"path": leaf.path, "version": leaf.version, "seq": seq})
    response["ETag"] = leaf.etag
    return response


@require_http_methods(["GET", "POST"])
def revisions(request, project_id):
    """Mark a version: one snapshot of the authored layer, with a summary."""
    try:
        project = Project.objects.get(id=project_id)
    except Project.DoesNotExist:
        return JsonResponse({"error": "no such project"}, status=404)
    if request.method == "GET":
        return JsonResponse(
            {
                "revisions": [
                    {"seq": r.seq, "author": r.author, "summary": r.summary,
                     "created": r.created.isoformat(), "leaves": len(r.document)}
                    for r in project.revisions.all()[:100]
                ]
            }
        )
    data = json.loads(request.body or b"{}")
    revision = Revision.objects.create(
        project=project,
        seq=project.seq,
        author=data.get("author") or "",
        summary=data.get("summary") or "",
        document={leaf.path: leaf.value for leaf in project.leaves.all()},
    )
    return JsonResponse({"seq": revision.seq, "leaves": len(revision.document)}, status=201)