feat(products): hosted-image URL build/parse helpers (SD-0002 §6.3.1)

DB-free host-agnostic helpers: image_url() builds the canonical hosted
Image Src for a fetched image; parse_image_id() recognizes one on re-import
via path-parse so exports round-trip across localhost/PPE/rebrand (INV-12).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app/domains/products/hosted.py

@@ -0,0 +1,30 @@
+"""Hosted-image URL helpers (SD-0002 §6.3.1, INV-12).
+
+Build the canonical hosted Image Src for a fetched image, and recognize one on
+re-import. DB-free and host-agnostic: recognition is a path-parse, so an export
+round-trips across localhost / PPE / the #23 rebrand. The diff resolves a parsed
+id against the storefront-scoped catalog snapshot.
+"""
+from __future__ import annotations
+
+import re
+from urllib.parse import urlsplit
+
+RENDITIONS = ("original", "thumb", "card", "detail")
+EXPORT_RENDITION = "detail"
+
+_PATH = re.compile(r"^/api/products/images/(\d+)/(original|thumb|card|detail)$")
+
+
+def image_url(base_url: str, image_id: int, rendition: str = EXPORT_RENDITION) -> str:
+    """Absolute when base_url is set, else a root-relative path."""
+    path = f"/api/products/images/{image_id}/{rendition}"
+    return f"{base_url.rstrip('/')}{path}" if base_url else path
+
+
+def parse_image_id(url: str) -> int | None:
+    """The image id if url is one of our hosted-image URLs (any host), else None."""
+    if not url:
+        return None
+    m = _PATH.match(urlsplit(url).path)
+    return int(m.group(1)) if m else None

backend/app/platform/config.py

@@ -68,3 +68,31 @@ def smtp_from() -> str:
 def smtp_starttls() -> bool:
     """STARTTLS on the relay connection (default on; disable only for odd relays)."""
     return os.environ.get("ECOMM_SMTP_STARTTLS", "1").strip().lower() not in {"0", "false", "no", "off"}
+
+
+def objectstore_kind() -> str:
+    """Which objectstore adapter to build: 'local' (dev/tests) or 'gcs' (deployed)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_KIND") or "local"
+
+
+def objectstore_bucket() -> str:
+    """The GCS bucket name for the media objects (gcs adapter; deployment overlay)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_BUCKET", "")
+
+
+def objectstore_local_dir() -> str:
+    """Base directory for the local-disk objectstore adapter (dev/tests)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_DIR") or "/tmp/ecomm-objectstore"
+
+
+def public_base_url() -> str:
+    """Absolute origin for hosted image URLs in export (e.g. APP_URL); '' → relative."""
+    return (os.environ.get("APP_URL") or "").rstrip("/")
+
+
+def image_fetch_allow_private() -> bool:
+    """Allow image fetch from private/loopback hosts (dev/E2E fixture host only).
+    Default off — the SSRF guard rejects private ranges in deployed envs."""
+    return os.environ.get("ECOMM_IMAGE_FETCH_ALLOW_PRIVATE", "").strip().lower() in {
+        "1", "true", "yes", "on",
+    }

backend/app/platform/images.py

@@ -0,0 +1,77 @@
+"""platform/images — pure image processing (SD-0002 §6.2).
+
+Model-free, deterministic, no I/O of its own: decode bytes, enforce the
+resolution bar (INV-18 / Q-3), and emit downscale-only WebP renditions. The
+caller (the image-fetch phase) owns fetching, storage, and status.
+"""
+from __future__ import annotations
+
+import io
+from dataclasses import dataclass
+
+from PIL import Image, UnidentifiedImageError
+
+# Q-3 (SD-0002 §13): the minimum shorter-side dimension; below it an image is
+# rejected_low_res. Rejects icons/thumbnails, accepts normal product photos.
+MIN_IMAGE_SHORT_SIDE = 500
+
+# Rendition longest-side caps (px); downscale-only — a smaller source is kept
+# at its own size, never upscaled. Encoded WebP.
+RENDITIONS = {"thumb": 160, "card": 480, "detail": 1600}
+
+# Source formats we accept (Pillow format names).
+_ACCEPTED = {"JPEG", "PNG", "WEBP"}
+
+_WEBP_QUALITY = 82
+
+
+@dataclass(frozen=True)
+class Processed:
+    source_format: str
+    width: int
+    height: int
+    renditions: dict[str, bytes]  # name -> WebP bytes
+
+
+@dataclass(frozen=True)
+class Rejected:
+    reason: str  # "rejected_low_res" | "rejected_not_image"
+
+
+def process(data: bytes) -> Processed | Rejected:
+    """Decode + bar-check + renditions, or a typed rejection. Never raises on
+    bad input — undecodable/unsupported bytes are Rejected('rejected_not_image')."""
+    try:
+        with Image.open(io.BytesIO(data)) as im:
+            im.load()
+            source_format = im.format or ""
+            if source_format not in _ACCEPTED:
+                return Rejected(reason="rejected_not_image")
+            rgb = im.convert("RGB")
+            width, height = rgb.size
+            if min(width, height) < MIN_IMAGE_SHORT_SIDE:
+                return Rejected(reason="rejected_low_res")
+            renditions = {
+                name: _rendition(rgb, cap) for name, cap in RENDITIONS.items()
+            }
+            return Processed(
+                source_format=source_format,
+                width=width,
+                height=height,
+                renditions=renditions,
+            )
+    except (UnidentifiedImageError, OSError, ValueError):
+        return Rejected(reason="rejected_not_image")
+
+
+def _rendition(rgb: Image.Image, cap: int) -> bytes:
+    longest = max(rgb.size)
+    if longest > cap:
+        scale = cap / longest
+        size = (max(1, round(rgb.width * scale)), max(1, round(rgb.height * scale)))
+        resized = rgb.resize(size, Image.LANCZOS)
+    else:
+        resized = rgb
+    buf = io.BytesIO()
+    resized.save(buf, format="WEBP", quality=_WEBP_QUALITY, method=6)
+    return buf.getvalue()

backend/app/platform/objectstore.py

@@ -0,0 +1,103 @@
+"""platform/objectstore — the media-blob port (SD-0002 §6.2, §6.3.1).
+
+A tiny put/get/delete port with two adapters, mirroring the SD-0001 mailer
+pattern: local-disk (dev/tests) and GCS (deployed). Owns no semantics — keys
+and content types come from the caller (the image-fetch phase). Objects are
+written once, never mutated (immutable, image-id-addressed).
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Protocol
+
+from app.platform import config
+
+
+class ObjectNotFound(Exception):
+    """Raised by get() when the key has no object."""
+
+
+class ObjectStore(Protocol):
+    def put(self, key: str, data: bytes, content_type: str) -> None: ...
+    def get(self, key: str) -> bytes: ...
+    def delete(self, key: str) -> None: ...
+
+
+def build_objectstore(kind: str) -> ObjectStore:
+    """Select the objectstore adapter by configured kind (config.objectstore_kind, INV-8)."""
+    if kind == "local":
+        return LocalObjectStore(config.objectstore_local_dir())
+    if kind == "gcs":
+        return GcsObjectStore(config.objectstore_bucket())
+    raise ValueError(f"unknown objectstore kind: {kind!r}")
+
+
+def _safe_segments(key: str) -> tuple[str, ...]:
+    parts = tuple(p for p in key.split("/") if p)
+    if not parts or any(p in {".", ".."} for p in parts):
+        raise ValueError(f"unsafe objectstore key: {key!r}")
+    return parts
+
+
+class LocalObjectStore:
+    """Disk-backed adapter under a base dir; key path-segments become subdirs."""
+
+    def __init__(self, base_dir: str) -> None:
+        self._base = Path(base_dir)
+
+    def _path(self, key: str) -> Path:
+        return self._base.joinpath(*_safe_segments(key))
+
+    def put(self, key: str, data: bytes, content_type: str) -> None:
+        path = self._path(key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    def get(self, key: str) -> bytes:
+        path = self._path(key)
+        try:
+            return path.read_bytes()
+        except FileNotFoundError as exc:
+            raise ObjectNotFound(key) from exc
+
+    def delete(self, key: str) -> None:
+        try:
+            os.remove(self._path(key))
+        except FileNotFoundError:
+            pass
+
+
+class GcsObjectStore:
+    """GCS adapter; auth via the VM service account ADC (no secret bytes).
+
+    google-cloud-storage is imported lazily so dev/test runs (local adapter)
+    don't require the package at import time.
+    """
+
+    def __init__(self, bucket_name: str) -> None:
+        if not bucket_name:
+            raise ValueError("gcs objectstore requires ECOMM_OBJECTSTORE_BUCKET")
+        from google.cloud import storage  # lazy
+
+        self._bucket = storage.Client().bucket(bucket_name)
+
+    def put(self, key: str, data: bytes, content_type: str) -> None:
+        _safe_segments(key)
+        self._bucket.blob(key).upload_from_string(data, content_type=content_type)
+
+    def get(self, key: str) -> bytes:
+        from google.cloud.exceptions import NotFound  # lazy
+
+        try:
+            return self._bucket.blob(key).download_as_bytes()
+        except NotFound as exc:
+            raise ObjectNotFound(key) from exc
+
+    def delete(self, key: str) -> None:
+        from google.cloud.exceptions import NotFound  # lazy
+
+        try:
+            self._bucket.blob(key).delete()
+        except NotFound:
+            pass

backend/requirements.txt

@@ -7,3 +7,5 @@ pytest>=8.0
 import-linter>=2.0
 nh3>=0.2
 python-multipart>=0.0.9
+Pillow>=11.0
+google-cloud-storage>=2.18

backend/tests/test_platform_images.py

@@ -0,0 +1,55 @@
+"""platform/images — pure decode + resolution bar + WebP renditions (SD-0002 §6.2)."""
+import io
+
+from PIL import Image
+
+from app.platform import images
+
+
+def _png(width: int, height: int, color=(120, 80, 200)) -> bytes:
+    buf = io.BytesIO()
+    Image.new("RGB", (width, height), color).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+def _jpeg(width: int, height: int) -> bytes:
+    buf = io.BytesIO()
+    Image.new("RGB", (width, height), (10, 200, 90)).save(buf, format="JPEG")
+    return buf.getvalue()
+
+
+def test_good_image_produces_three_webp_renditions():
+    result = images.process(_png(1200, 900))
+    assert isinstance(result, images.Processed)
+    assert result.source_format == "PNG"
+    assert set(result.renditions) == {"thumb", "card", "detail"}
+    for name, blob in result.renditions.items():
+        with Image.open(io.BytesIO(blob)) as im:
+            assert im.format == "WEBP"
+    with Image.open(io.BytesIO(result.renditions["thumb"])) as im:
+        assert max(im.size) <= images.RENDITIONS["thumb"]
+    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
+        assert max(im.size) <= images.RENDITIONS["detail"]
+
+
+def test_downscale_only_never_upscales_small_source():
+    result = images.process(_png(520, 520))
+    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
+        assert im.size == (520, 520)
+
+
+def test_below_resolution_bar_rejected_low_res():
+    result = images.process(_png(300, 1200))  # shorter side 300 < 500
+    assert isinstance(result, images.Rejected)
+    assert result.reason == "rejected_low_res"
+
+
+def test_not_an_image_rejected_not_image():
+    result = images.process(b"this is not an image")
+    assert isinstance(result, images.Rejected)
+    assert result.reason == "rejected_not_image"
+
+
+def test_jpeg_source_format_preserved_in_result():
+    result = images.process(_jpeg(800, 800))
+    assert result.source_format == "JPEG"

backend/tests/test_platform_objectstore.py

@@ -0,0 +1,44 @@
+"""platform/objectstore — local-disk adapter round-trip (SD-0002 §6.2/§6.3.1)."""
+import pytest
+
+from app.platform import objectstore
+
+
+def test_local_put_get_round_trip(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    key = "storefronts/1/product-images/9/original"
+    store.put(key, b"\x89PNG-bytes", "image/png")
+    assert store.get(key) == b"\x89PNG-bytes"
+
+
+def test_local_get_missing_raises_not_found(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    with pytest.raises(objectstore.ObjectNotFound):
+        store.get("nope/missing")
+
+
+def test_local_delete_is_idempotent(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    store.put("a/b", b"x", "text/plain")
+    store.delete("a/b")
+    store.delete("a/b")  # no error second time
+    with pytest.raises(objectstore.ObjectNotFound):
+        store.get("a/b")
+
+
+def test_local_keys_cannot_escape_base_dir(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    with pytest.raises(ValueError):
+        store.put("../escape", b"x", "text/plain")
+
+
+def test_build_objectstore_local(tmp_path, monkeypatch):
+    monkeypatch.setenv("ECOMM_OBJECTSTORE_KIND", "local")
+    monkeypatch.setenv("ECOMM_OBJECTSTORE_DIR", str(tmp_path))
+    store = objectstore.build_objectstore("local")
+    assert isinstance(store, objectstore.LocalObjectStore)
+
+
+def test_build_objectstore_unknown_kind_raises():
+    with pytest.raises(ValueError):
+        objectstore.build_objectstore("s3")

backend/tests/test_products_hosted.py

@@ -0,0 +1,30 @@
+"""hosted-image URL helpers — round-trip recognition (SD-0002 §6.3.1, INV-12)."""
+from app.domains.products import hosted
+
+
+def test_build_relative_when_no_base():
+    assert hosted.image_url("", 42, "detail") == "/api/products/images/42/detail"
+
+
+def test_build_absolute_with_base():
+    url = hosted.image_url("https://ecomm-ppe.wiggleverse.org", 42, "detail")
+    assert url == "https://ecomm-ppe.wiggleverse.org/api/products/images/42/detail"
+
+
+def test_parse_absolute_hosted_url():
+    url = "https://market.wiggleverse.org/api/products/images/7/detail"
+    assert hosted.parse_image_id(url) == 7
+
+
+def test_parse_relative_hosted_url():
+    assert hosted.parse_image_id("/api/products/images/7/card") == 7
+
+
+def test_parse_ignores_query_and_trailing():
+    assert hosted.parse_image_id("/api/products/images/7/detail?v=1") == 7
+
+
+def test_parse_non_hosted_returns_none():
+    assert hosted.parse_image_id("https://cdn.example.com/a/b.png") is None
+    assert hosted.parse_image_id("/api/products/images/abc/detail") is None
+    assert hosted.parse_image_id("/api/products/images/7") is None