backend/app/domains/products/__init__.py

@@ -20,6 +20,7 @@ from .errors import (
     RunNotFound,
 )
 from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
+from .repo import image_for_serving
 from .service import (
     confirm_draft,
     discard_draft,
@@ -41,5 +42,5 @@ __all__ = [
     "MAX_DATA_ROWS", "MAX_FILE_BYTES", "SAMPLE_CSV_PATH",
     "import_validate", "get_draft", "get_draft_records", "discard_draft",
     "confirm_draft", "list_runs", "get_run", "summary", "export_catalog",
-    "run_image_phase", "recover_incomplete_runs",
+    "run_image_phase", "recover_incomplete_runs", "image_for_serving",
 ]

backend/app/domains/products/repo.py

@@ -453,6 +453,19 @@ def incomplete_runs(conn: psycopg.Connection) -> list[dict]:
     return [{"id": r[0], "storefront_id": r[1]} for r in rows]
 
 
+def image_for_serving(conn: psycopg.Connection, storefront_id: int, image_id: int) -> dict | None:
+    """Storefront-scoped lookup for the serving route: status + rendition keys (INV-14)."""
+    row = conn.execute(
+        "SELECT i.status, i.key_original, i.key_thumb, i.key_card, i.key_detail"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " WHERE i.id = %s AND p.storefront_id = %s",
+        (image_id, storefront_id),
+    ).fetchone()
+    if row is None:
+        return None
+    return {"status": row[0], "original": row[1], "thumb": row[2], "card": row[3], "detail": row[4]}
+
+
 def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]:
     rows = conn.execute(
         "SELECT p.handle, i.source_url, i.status, i.failure_reason,"

backend/app/main.py

@@ -18,6 +18,7 @@ from typing import Any
 
 import psycopg
 from fastapi import Depends, FastAPI, File, Query, Response, UploadFile
+from fastapi import Path as ApiPath
 from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
@@ -410,6 +411,36 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
             headers={"content-disposition": 'attachment; filename="ecomm-products-export.csv"'},
         )
 
+    _RENDITION_CT = {"thumb": "image/webp", "card": "image/webp",
+                     "detail": "image/webp", "original": "application/octet-stream"}
+
+    @app.get("/api/products/images/{image_id}/{rendition}")
+    def serve_product_image(
+        image_id: int,
+        rendition: str = ApiPath(pattern="^(original|thumb|card|detail)$"),
+        conn: psycopg.Connection = Depends(get_conn),
+        sess: dict | None = Depends(get_session),
+    ):
+        """Serve a hosted image rendition, storefront-authorized + immutable cache (§6.4, INV-16)."""
+        gate = _merchant_gate(conn, sess)
+        if isinstance(gate, JSONResponse):
+            return gate
+        _account, sf = gate
+        rec = products.image_for_serving(conn, sf.id, image_id)
+        if rec is None:
+            return _error(404, "not_found", "No such image.")
+        if rec["status"] != "fetched":
+            return _error(409, "not_fetched", "This image has not been fetched yet.")
+        key = rec[rendition]
+        if not key:
+            return _error(404, "not_found", "No such rendition.")
+        try:
+            data = app.state.objectstore.get(key)
+        except objectstore_mod.ObjectNotFound:
+            return _error(404, "not_found", "No such image.")
+        return Response(content=data, media_type=_RENDITION_CT[rendition],
+                        headers={"Cache-Control": "public, max-age=31536000, immutable"})
+
     @app.get("/api/products/sample.csv")
     def products_sample_csv():
         """The DOC-3 worked-example CSV. Documentation, so no auth gate (§6.4)."""

backend/tests/test_products_image_serving.py

@@ -0,0 +1,78 @@
+"""GET /api/products/images/{id}/{rendition} — authorize, stream, immutable cache (§6.4, INV-14/16)."""
+import psycopg
+from contextlib import contextmanager
+
+from fastapi.testclient import TestClient
+
+from app.main import create_app
+from app.domains import products  # noqa: F401  (ensures package import path)
+
+from test_products_endpoints import _merchant_client  # reuse the signed-in client
+
+
+def _seed_image(fresh_db_url, status="fetched", with_detail_bytes=None, store=None,
+                email_storefront_only=True):
+    """Insert a product + image for the (only) storefront; set keys; return (storefront_id, image_id, keys)."""
+    with psycopg.connect(fresh_db_url) as conn:
+        sf = conn.execute("SELECT id FROM storefront ORDER BY id LIMIT 1").fetchone()[0]
+        pid = conn.execute(
+            "INSERT INTO product (storefront_id, handle, title) VALUES (%s,'lamp','Lamp') RETURNING id",
+            (sf,)).fetchone()[0]
+        iid = conn.execute(
+            "INSERT INTO product_image (product_id, source_url, position, status)"
+            " VALUES (%s,'https://m/a.png',1,%s) RETURNING id", (pid, status)).fetchone()[0]
+        keys = {r: f"storefronts/{sf}/product-images/{iid}/{r}" for r in ("original", "thumb", "card", "detail")}
+        if status == "fetched":
+            conn.execute(
+                "UPDATE product_image SET key_original=%s, key_thumb=%s, key_card=%s, key_detail=%s WHERE id=%s",
+                (keys["original"], keys["thumb"], keys["card"], keys["detail"], iid))
+        conn.commit()
+    if with_detail_bytes is not None and store is not None:
+        store.put(keys["detail"], with_detail_bytes, "image/webp")
+    return sf, iid, keys
+
+
+def test_serves_fetched_rendition_with_immutable_cache(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="fetched",
+                                      with_detail_bytes=b"WEBPDATA", store=client.app.state.objectstore)
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 200
+        assert resp.headers["content-type"] == "image/webp"
+        assert "immutable" in resp.headers.get("cache-control", "")
+        assert resp.content == b"WEBPDATA"
+
+
+def test_not_fetched_returns_409(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="pending")
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 409
+        assert resp.json()["error"]["code"] == "not_fetched"
+
+
+def test_other_storefronts_image_is_404(fresh_db_url):
+    # Sign in as merchant B first (this migrates the DB), then seed an image
+    # belonging to a *different* storefront A — it must be invisible to B.
+    with _merchant_client(fresh_db_url, email="b@example.com") as client:
+        with psycopg.connect(fresh_db_url) as conn:
+            a = conn.execute("INSERT INTO account (email) VALUES ('a@example.com') RETURNING id").fetchone()[0]
+            sfa = conn.execute("INSERT INTO storefront (name) VALUES ('A') RETURNING id").fetchone()[0]
+            conn.execute("INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (a, sfa))
+            pid = conn.execute("INSERT INTO product (storefront_id, handle, title) VALUES (%s,'lamp','Lamp') RETURNING id", (sfa,)).fetchone()[0]
+            iid = conn.execute("INSERT INTO product_image (product_id, source_url, position, status) VALUES (%s,'u',1,'fetched') RETURNING id", (pid,)).fetchone()[0]
+            conn.commit()
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 404
+
+
+def test_unauthenticated_is_401(fresh_db_url):
+    with TestClient(create_app(database_url=fresh_db_url)) as client:
+        assert client.get("/api/products/images/1/detail").status_code == 401
+
+
+def test_bad_rendition_is_422(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="fetched",
+                                      with_detail_bytes=b"x", store=client.app.state.objectstore)
+        assert client.get(f"/api/products/images/{iid}/huge").status_code == 422