From d7b8e8c46c3029b6a08a257c5ec460555edc0924 Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Fri, 12 Jun 2026 01:22:56 -0700 Subject: [PATCH] add plan ./plans/2026-06-11-slice-7-images.md --- plans/2026-06-11-slice-7-images.md | 1933 ++++++++++++++++++++++++++++ 1 file changed, 1933 insertions(+) create mode 100644 plans/2026-06-11-slice-7-images.md diff --git a/plans/2026-06-11-slice-7-images.md b/plans/2026-06-11-slice-7-images.md new file mode 100644 index 0000000..dbd2af6 --- /dev/null +++ b/plans/2026-06-11-slice-7-images.md @@ -0,0 +1,1933 @@ +# SLICE-7 — Images pipeline end-to-end (SD-0002 §7.2) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fetch, validate, host, and serve merchant product images — closing PUC-7 — so imported catalogs land with hosted renditions, image outcomes are reported per image, and the export → re-import round-trip stays a no-op over hosted image URLs (INV-12 extended over images). + +**Architecture:** Two new model-free platform modules (`platform/images` = decode + resolution bar + WebP renditions; `platform/objectstore` = a `put/get/delete` port with local-disk and GCS adapters, mirroring the SD-0001 mailer port). A post-commit, in-process **image-fetch phase** in the products domain (`domains/products/imagefetch.py`) walks a run's `pending` images with bounded concurrency (~4 worker threads), fetches each URL behind an SSRF guard with INV-18 bounds, runs `platform/images`, stores renditions via `objectstore`, and moves each image `pending → fetched | rejected_* | failed` — idempotent per image, so it resumes after a restart. An app-served, storefront-authorized route streams renditions from the bucket with immutable caching (INV-16); exported `Image Src` carries the app's own hosted URL for fetched images and the original `source_url` otherwise, and the importer recognizes a hosted URL by path-parse and resolves it to the existing image record (never re-fetched), keeping the round-trip lossless. The per-environment private media bucket is provisioned by a new flotilla/launch-app `provision-bucket` operator gesture at the PPE deploy. + +**Tech Stack:** Python 3 / FastAPI / psycopg 3 (sync) · Pillow (decode + WebP renditions) · httpx (sync, SSRF-guarded fetch) · `concurrent.futures.ThreadPoolExecutor` (bounded per-image concurrency) · google-cloud-storage (GCS adapter; VM service-account ADC) · React/Vite SPA (polling run detail) · Playwright E2E (local fixture image host) · launch-app skill + `gcloud storage` (bucket provisioning). + +**Anchor:** SD-0002 `wiggleverse-ecomm-content/specs/SD-0002-products-bulk-csv-import-export.md` §7.2 SLICE-7 (graduated Solution Design, R2a) · issue wiggleverse/wiggleverse-ecomm#13. + +--- + +## Decisions locked for this slice (read before starting) + +- **Q-3 resolution bar = shorter side ≥ 500 px.** Below it → `rejected_low_res`. A single constant `MIN_IMAGE_SHORT_SIDE = 500` in `platform/images.py`; rationale recorded in SD-0002 §13. Consistent with the downscale-only rendition ladder (detail ≤1600 / card ~480 / thumb ~160 never upscale a ≥500 source). +- **Fetch phase = background thread + inner `ThreadPoolExecutor(max_workers=4)`**, not asyncio. The stack is sync (sync psycopg pool, sync httpx); a thread runner satisfies the binding invariants (in-process, post-commit, ~4 concurrent, resumable, per-image idempotent) without an async/sync DB bridge. The §6.9 queue-portability seam is unchanged. +- **Draft upload blobs stay BYTEA this slice.** The §7.2 *Ships* list is images-only; moving draft blobs to the bucket (the softer `products-domain.md` seam) is deferred to a later hardening pass. The media bucket holds only `product-images/` objects this slice. (The `provision-bucket` script still sets the `import-drafts/` lifecycle rule for forward-compat — harmless when no such objects exist.) +- **No new migration.** `0002_products.sql` already carries every needed column: `product_image.status` (enum incl. `fetched`/`rejected_low_res`/`rejected_not_image`/`failed`), `failure_reason`, `key_original`/`key_thumb`/`key_card`/`key_detail`, `import_run_id`, `fetched_at`; `import_run.status` enum incl. `fetching_images`/`complete_with_problems`. +- **Hosted-URL recognition is host-agnostic and DB-free.** A hosted URL is any URL whose *path* matches `/api/products/images/{image_id}/{rendition}`. Recognition is a pure path-parse (survives the #23 rebrand and works across localhost/PPE). The diff resolves a recognized id against the (already storefront-scoped) catalog snapshot — an id not in this storefront's snapshot is a row error, never a fetch. +- **Public base URL for export = `APP_URL`** (already in `deployment.toml [overlay]`). When unset (local/dev), the serializer emits a relative `/api/products/images/{id}/detail` path — still recognized on re-import. + +--- + +## File structure + +**Backend — new files** +- `backend/app/platform/images.py` — pure: decode (jpeg/png/webp), resolution bar, WebP renditions. No I/O. +- `backend/app/platform/objectstore.py` — `ObjectStore` Protocol + `LocalObjectStore` + `GcsObjectStore` + `build_objectstore(kind)`. +- `backend/app/domains/products/hosted.py` — hosted-image URL build/parse helpers (DB-free). +- `backend/app/domains/products/imagefetch.py` — the image-fetch phase + SSRF guard + startup recovery (uses repo + platform/images + platform/objectstore). +- `backend/tests/test_platform_images.py`, `backend/tests/test_platform_objectstore.py`, `backend/tests/test_products_hosted.py`, `backend/tests/test_products_imagefetch.py`, `backend/tests/test_products_image_serving.py`. + +**Backend — modified** +- `backend/app/platform/config.py` — `objectstore_kind`, `objectstore_bucket`, `objectstore_local_dir`, `public_base_url`, `image_fetch_allow_private`. +- `backend/app/domains/products/diff.py` — `CatalogImage` gains `status`; image matching resolves hosted URLs. +- `backend/app/domains/products/repo.py` — `load_catalog` loads `status` + `key_detail`; image-phase repo helpers; `get_run`/`list_runs` image progress/outcomes/counts. +- `backend/app/domains/products/serialize.py` — `_write_image` emits hosted detail URL for fetched images. +- `backend/app/domains/products/service.py` — `confirm_draft` sets `fetching_images` when the run has pending images; `export_catalog` passes the base URL; serialize signature threaded. +- `backend/app/domains/products/__init__.py` — export the new image-phase entrypoints. +- `backend/app/main.py` — build objectstore in lifespan; startup recovery scan; `app.state.image_runner`; schedule the fetch phase post-confirm; `GET /api/products/images/{image_id}/{rendition}`. +- `backend/requirements.txt` — add `Pillow`, `google-cloud-storage`. +- existing `backend/tests/test_products_serialize.py`, `test_products_diff.py`, `test_products_service.py` — extend for hosted images. + +**Frontend — modified** +- `frontend/src/productsApi.ts` — typed `ImageOutcome`, image counts, image-serving URL builder. +- `frontend/src/screens/products/RunDetail.tsx` — images section (live progress + outcomes table; poll while non-terminal). +- `frontend/src/screens/products/ProductsPage.tsx` — image-problems notice band; history "image outcomes" column. +- `frontend/src/screens/products/runImages.ts` (new) — pure helpers (progress label, outcome label) + vitest. + +**E2E — new/modified** +- `e2e/fixtures/images/` (good.png 800×800, tiny.png 64×64, served by a fixture host), `e2e/fixtures/with-images.csv`, `e2e/image-host.mjs` (or python), `e2e/serve.sh` (start fixture host + set env), `e2e/tests/image-outcomes.spec.ts`, `e2e/helpers.ts` (importWithImages helper). + +**Cross-repo (engineering) — new** +- `engineering/launch-app/skills/provision-bucket/SKILL.md` + `assets/provision-bucket.sh` + `assets/read` reuse; `engineering/launch-app/SPEC.md` §6.6b note. + +**Deploy/docs — modified** +- `deployment.toml` `[overlay]` — `ECOMM_OBJECTSTORE_KIND`, `ECOMM_OBJECTSTORE_BUCKET`. +- `docs/products-domain.md`, `docs/OPERATIONS.md`, `VERSION`, `frontend/package.json` (+lockfile). +- SD-0002 §13 Q-3 marked resolved (content repo, at finalize alongside plan archive). + +--- + +## Task 1: `platform/images` — decode, resolution bar, WebP renditions + +**Files:** +- Create: `backend/app/platform/images.py` +- Create: `backend/tests/test_platform_images.py` +- Modify: `backend/requirements.txt` + +- [ ] **Step 1: Add Pillow to requirements** + +Add to `backend/requirements.txt` (keep alphabetical-ish grouping with the other runtime deps): + +``` +Pillow>=11.0 +``` + +Then install into the worktree venv: `./.venv/bin/pip install -r backend/requirements.txt` (the worktree's venv is created in Task 0 of execution / by `scripts/dev.sh`; if absent, run `python3 -m venv .venv && ./.venv/bin/pip install -r backend/requirements.txt`). + +- [ ] **Step 2: Write the failing tests** + +`backend/tests/test_platform_images.py`: + +```python +"""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" + # downscale-only: longest side never exceeds the rendition cap + 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(): + # 520x520 passes the bar; detail cap is 1600 but must not upscale. + 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" +``` + +- [ ] **Step 3: Run to verify failure** + +Run: `./.venv/bin/python -m pytest backend/tests/test_platform_images.py -q` +Expected: FAIL — `ModuleNotFoundError: app.platform.images`. + +- [ ] **Step 4: Implement `platform/images.py`** + +```python +"""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() +``` + +- [ ] **Step 5: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_platform_images.py -q` +Expected: PASS (5 tests). + +- [ ] **Step 6: Commit** + +```bash +git add backend/app/platform/images.py backend/tests/test_platform_images.py backend/requirements.txt +git commit -m "feat(platform): images port — decode, resolution bar (Q-3), WebP renditions (SD-0002 §6.2)" +``` + +--- + +## Task 2: `platform/objectstore` — port + local & GCS adapters + +**Files:** +- Create: `backend/app/platform/objectstore.py` +- Create: `backend/tests/test_platform_objectstore.py` +- Modify: `backend/app/platform/config.py`, `backend/requirements.txt` + +- [ ] **Step 1: Add config accessors** + +Append to `backend/app/platform/config.py` (mirror the existing `os.environ.get` style): + +```python +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", + } +``` + +- [ ] **Step 2: Add google-cloud-storage to requirements** + +Add to `backend/requirements.txt`: + +``` +google-cloud-storage>=2.18 +``` + +Install: `./.venv/bin/pip install -r backend/requirements.txt`. + +- [ ] **Step 3: Write the failing tests (local adapter)** + +`backend/tests/test_platform_objectstore.py`: + +```python +"""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") +``` + +- [ ] **Step 4: Run to verify failure** + +Run: `./.venv/bin/python -m pytest backend/tests/test_platform_objectstore.py -q` +Expected: FAIL — module missing. + +- [ ] **Step 5: Implement `platform/objectstore.py`** + +```python +"""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 +``` + +- [ ] **Step 6: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_platform_objectstore.py -q` +Expected: PASS (6 tests). + +- [ ] **Step 7: Verify layering still holds** + +Run: `cd backend && ../.venv/bin/lint-imports` (the `main > domains > platform` contract; `platform` imports only stdlib/third-party). Expected: contract kept. + +- [ ] **Step 8: Commit** + +```bash +git add backend/app/platform/objectstore.py backend/app/platform/config.py backend/tests/test_platform_objectstore.py backend/requirements.txt +git commit -m "feat(platform): objectstore port — local + GCS adapters, config (SD-0002 §6.2)" +``` + +--- + +## Task 3: hosted-image URL build/parse helpers + +**Files:** +- Create: `backend/app/domains/products/hosted.py` +- Create: `backend/tests/test_products_hosted.py` + +The contract: build a hosted `Image Src` for a fetched image; recognize one on re-import. Path is `/api/products/images/{image_id}/{rendition}` (§6.4); recognition is host-agnostic (path-parse). + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_products_hosted.py`: + +```python +"""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 +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_hosted.py -q` +Expected: FAIL — module missing. + +- [ ] **Step 3: Implement `hosted.py`** + +```python +"""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 +``` + +- [ ] **Step 4: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_hosted.py -q` +Expected: PASS (6 tests). + +- [ ] **Step 5: Commit** + +```bash +git add backend/app/domains/products/hosted.py backend/tests/test_products_hosted.py +git commit -m "feat(products): hosted-image URL build/parse helpers (SD-0002 §6.3.1)" +``` + +--- + +## Task 4: catalog snapshot carries image status; serializer emits hosted URLs + +**Files:** +- Modify: `backend/app/domains/products/diff.py` (CatalogImage dataclass) +- Modify: `backend/app/domains/products/repo.py` (`load_catalog` image query) +- Modify: `backend/app/domains/products/serialize.py` (`_write_image`, `catalog_to_csv` signature) +- Modify: `backend/app/domains/products/service.py` (`export_catalog` passes base url) +- Modify: `backend/tests/test_products_serialize.py` + +- [ ] **Step 1: Write the failing test (serializer emits hosted URL for fetched, source for others)** + +Add to `backend/tests/test_products_serialize.py` (reuse its existing `CatalogProduct`/`CatalogImage`/`CatalogVariant` builders; pass `status` to images): + +```python +def test_fetched_image_serializes_hosted_detail_url(): + from app.domains.products import serialize + from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant + + product = CatalogProduct( + id=1, handle="lamp", title="Lamp", option_names=(None, None, None), + fields={"title": "Lamp", "status": "active"}, + variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={})], + images=[ + CatalogImage(id=55, source_url="https://m.example/a.png", position=1, + alt_text=None, status="fetched"), + CatalogImage(id=56, source_url="https://m.example/b.png", position=2, + alt_text=None, status="failed"), + ], + ) + rows = list(serialize._product_rows(product, base_url="https://shop.test")) + srcs = [r.get("Image Src") for r in rows if r.get("Image Src")] + assert "https://shop.test/api/products/images/55/detail" in srcs # fetched -> hosted + assert "https://m.example/b.png" in srcs # failed -> source kept +``` + +(Adjust the helper call to however the existing test constructs products; the key assertions are the two `Image Src` values.) + +- [ ] **Step 2: Run to verify failure** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q` +Expected: FAIL — `CatalogImage` has no `status`; `_product_rows`/`_write_image` take no `base_url`. + +- [ ] **Step 3: Extend `CatalogImage` in `diff.py`** + +In `backend/app/domains/products/diff.py`, add `status` (default keeps existing call sites valid): + +```python +@dataclass +class CatalogImage: + id: int + source_url: str + position: int + alt_text: str | None + status: str = "pending" +``` + +- [ ] **Step 4: Load `status` in `repo.load_catalog`** + +In `repo.py`, the image query — add `i.status` and pass it: + +```python + for row in conn.execute( + "SELECT i.product_id, i.id, i.source_url, i.position, i.alt_text, i.status" + " FROM product_image i" + " JOIN product p ON p.id = i.product_id" + " WHERE p.storefront_id = %s" + " ORDER BY i.product_id, i.position, i.id", + (storefront_id,), + ): + by_id[row[0]].images.append( + CatalogImage(id=row[1], source_url=row[2], position=row[3], + alt_text=row[4], status=row[5]) + ) +``` + +- [ ] **Step 5: Thread `base_url` through the serializer** + +In `serialize.py`: `catalog_to_csv(snapshot, base_url="")` → `_product_rows(product, base_url)` → `_write_image(row, image, base_url)`: + +```python +from .hosted import EXPORT_RENDITION, image_url + + +def _write_image(row: dict[str, str], image, base_url: str) -> None: + # Fetched images export the platform-hosted URL (INV-12/16); everything + # else keeps the original source_url so nothing is lost (§6.5.5). + if image.status == "fetched": + row["Image Src"] = image_url(base_url, image.id, EXPORT_RENDITION) + else: + row["Image Src"] = image.source_url + row["Image Position"] = str(image.position) + if image.alt_text is not None: + row["Image Alt Text"] = image.alt_text +``` + +Update `catalog_to_csv` and `_product_rows` signatures to take and forward `base_url` (default `""`). + +- [ ] **Step 6: Pass the base url from `service.export_catalog`** + +In `service.py`, import config and pass it into the stream: + +```python +from app.platform import config +... + yield from serialize.catalog_to_csv(snapshot, base_url=config.public_base_url()) +``` + +- [ ] **Step 7: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q` +Expected: PASS (incl. the new test). Then full export test: `./.venv/bin/python -m pytest backend/tests/test_products_export.py -q` (existing snapshots use never-fetched images → still emit source_url → unchanged). + +- [ ] **Step 8: Commit** + +```bash +git add backend/app/domains/products/diff.py backend/app/domains/products/repo.py backend/app/domains/products/serialize.py backend/app/domains/products/service.py backend/tests/test_products_serialize.py +git commit -m "feat(products): export hosted detail URL for fetched images; snapshot carries status (SD-0002 §6.5.5, INV-16)" +``` + +--- + +## Task 5: diff resolves hosted URLs (INV-12 over images) + +**Files:** +- Modify: `backend/app/domains/products/diff.py` (image matching in `_update_plan`; add-path hosted check) +- Modify: `backend/tests/test_products_diff.py` +- Modify: `backend/tests/test_products_serialize.py` (extend the INV-12 property test to fetched images) + +The rule: +1. A canonical image whose `source_url` parses as a hosted id (`hosted.parse_image_id`) is **not a new URL**. Resolve it to the catalog image with that id **within the same product**: matched → treat as the existing image (position/alt diff only; never an add → never a fetch). If the id is not a catalog image of this product (cross-storefront, deleted, or another product) → **row error** ("image refers to an unknown hosted id") — never a fetch attempt on our own domain. +2. A non-hosted (original) URL keeps today's behavior: match by `source_url`, else add. + +- [ ] **Step 1: Write the failing tests** + +Add to `backend/tests/test_products_diff.py`: + +```python +def test_hosted_url_for_existing_image_is_unchanged_not_add(): + # Catalog has a fetched image id=55; canonical re-imports it as the hosted URL. + from app.domains.products import diff + from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant + from app.domains.products.models import CanonicalImage, CanonicalProduct, CanonicalVariant + + catalog = { + "lamp": CatalogProduct( + id=1, handle="lamp", title="Lamp", option_names=(None, None, None), + fields={"title": "Lamp", "status": "active", "description_html": None, + "vendor": None, "product_type": "standalone", + "google_product_category": None, "tags": [], "published": True}, + variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={ + "sku": None, "barcode": None, "price": None, "cost": None, + "weight": None, "weight_unit": None, "volume": None, "volume_unit": None, + "tax_id_1": None, "tax_id_2": None, "inventory_tracker": None, + "inventory_qty": None, "variant_image": None})], + images=[CatalogImage(id=55, source_url="https://m.example/a.png", + position=1, alt_text=None, status="fetched")], + ) + } + canonical = [CanonicalProduct( + first_line=2, handle="lamp", title="Lamp", + variants=[CanonicalVariant(line_number=2, options=(None, None, None), fields={})], + images=[CanonicalImage(line_number=2, + source_url="/api/products/images/55/detail", + position=1, alt_text=None)], + )] + result = diff.compute_diff(catalog, canonical) + handle_kind = {r["handle"]: r["kind"] for r in result.records} + assert handle_kind["lamp"] == "unchanged" # not "update", not "add" + + +def test_hosted_url_for_unknown_id_is_row_error(): + from app.domains.products import diff + from app.domains.products.diff import CatalogProduct, CatalogVariant + from app.domains.products.models import CanonicalImage, CanonicalProduct, CanonicalVariant + + catalog = { + "lamp": CatalogProduct( + id=1, handle="lamp", title="Lamp", option_names=(None, None, None), + fields={"title": "Lamp", "status": "active", "description_html": None, + "vendor": None, "product_type": "standalone", + "google_product_category": None, "tags": [], "published": True}, + variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={ + "sku": None, "barcode": None, "price": None, "cost": None, + "weight": None, "weight_unit": None, "volume": None, "volume_unit": None, + "tax_id_1": None, "tax_id_2": None, "inventory_tracker": None, + "inventory_qty": None, "variant_image": None})], + images=[], + ) + } + canonical = [CanonicalProduct( + first_line=2, handle="lamp", title="Lamp", + variants=[CanonicalVariant(line_number=2, options=(None, None, None), fields={})], + images=[CanonicalImage(line_number=2, + source_url="/api/products/images/999/detail", + position=1, alt_text=None)], + )] + result = diff.compute_diff(catalog, canonical) + kinds = {r["handle"]: r["kind"] for r in result.records} + assert kinds["lamp"] == "error" +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_diff.py -k hosted -q` +Expected: FAIL — hosted URLs currently treated as a new add. + +- [ ] **Step 3: Implement hosted resolution in `_update_plan` (and the add path)** + +In `diff.py`, before the existing `by_src` matching loop in the update path, build an id index and resolve hosted canonical images. Concretely, in the image loop replace the match lookup: + +```python +from . import hosted +... + by_src = {i.source_url: i for i in current.images} + by_id = {i.id: i for i in current.images} + for image in product.images: + hosted_id = hosted.parse_image_id(image.source_url) + if hosted_id is not None: + match = by_id.get(hosted_id) + if match is None: + # Hosted URL whose id is not one of this product's images: + # unknown/cross-storefront — a row error, never a fetch. + product.errors.append( + RowError(image.line_number, "Image Src", + f"image '{image.source_url}' refers to an unknown hosted id") + ) + continue + # Resolve to the existing image; only position/alt can differ. A + # hosted re-import of the same image at the same slot is unchanged. + image = _resolved_hosted(image, match) # see helper below + else: + match = by_src.get(image.source_url) + if match is None: + # ... existing add branch unchanged ... + continue + # ... existing position/alt update-compare branch, using `match` ... +``` + +Add a small helper so the position/alt compare reuses the matched image's identity: + +```python +def _resolved_hosted(image, match): + """A hosted-URL canonical image, re-keyed onto the matched catalog image's + source_url so the downstream position/alt compare and apply behave as an + existing-image update (never an add).""" + return replace(image, source_url=match.source_url) +``` + +(`from dataclasses import replace`. `CanonicalImage` is frozen — `replace` returns a new instance.) For the **add path** (a non-hosted brand-new URL) nothing changes — it stays an add (→ pending → fetched by the task). + +Important: when the product is an **add** (`plan.kind == "add"`, no `current`), a hosted URL cannot resolve to any existing image → it must be a row error too. In the add branch, before emitting image add-plans, run the same `hosted.parse_image_id` check: a hosted URL on a brand-new product is an unknown id → row error (downgrades the product to error, consistent with the update path). Add a guard at the top of the add-image handling. + +- [ ] **Step 4: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_diff.py -q` +Expected: PASS (incl. the two new tests + all existing). + +- [ ] **Step 5: Extend the INV-12 property test over fetched images** + +In `backend/tests/test_products_serialize.py`, the existing 200-catalog property generator (`diff(catalog, import(export(catalog))) == ∅`) — extend it so a fraction of images have `status="fetched"` (with stable ids), build the export with a fixed `base_url="https://shop.test"`, and assert the re-import diffs to all-unchanged. The serializer emits hosted URLs for fetched images; the parse-back in `parse_csv`/`validate` produces canonical images carrying those hosted URLs; the diff resolves them to the same ids → unchanged. Concretely, where the generator creates each `CatalogImage`, set `status` to `"fetched"` for, say, every other image, and pass `base_url` into `serialize.catalog_to_csv(...)`. Keep `never-fetched` images too (they export source_url and round-trip as before). + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q` +Expected: PASS — round-trip no-op holds with fetched (hosted) and non-fetched images mixed. + +- [ ] **Step 6: Commit** + +```bash +git add backend/app/domains/products/diff.py backend/tests/test_products_diff.py backend/tests/test_products_serialize.py +git commit -m "feat(products): diff resolves hosted image URLs to existing records — INV-12 over images (SD-0002 §6.3)" +``` + +--- + +## Task 6: image-phase repo helpers + run progress/outcomes in payloads + +**Files:** +- Modify: `backend/app/domains/products/repo.py` +- Modify: `backend/tests/test_products_service.py` (or a new `test_products_repo_images.py`) + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_products_repo_images.py` (uses the existing pytest Postgres fixtures — copy the conftest connection fixture pattern other repo tests use; insert a storefront/product/run/images directly, then assert helper behavior): + +```python +"""products repo — image-phase helpers (SD-0002 §6.5.4).""" +# Use the project's existing DB test fixtures (see conftest.py): a connection +# bound to a migrated test database, plus helpers to seed an account/storefront. + +def test_pending_images_for_run_returns_only_pending(db, seed_storefront): + sf = seed_storefront(db) + run_id = _seed_run(db, sf, status="fetching_images") + p = _seed_product(db, sf, "lamp") + a = _seed_image(db, p, "https://m/a.png", run_id, status="pending") + _seed_image(db, p, "https://m/b.png", run_id, status="fetched") + pend = repo.pending_images_for_run(db, run_id) + assert [img["id"] for img in pend] == [a] + assert pend[0]["source_url"] == "https://m/a.png" + assert pend[0]["storefront_id"] == sf + + +def test_claim_image_for_fetch_is_idempotent(db, seed_storefront): + sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images") + p = _seed_product(db, sf, "lamp") + img = _seed_image(db, p, "https://m/a.png", run_id, status="pending") + assert repo.claim_image_for_fetch(db, img) is True + # second claim returns False (already claimed/terminal) — restart safety + repo.mark_image_fetched(db, img, {"original": "k/o", "thumb": "k/t", + "card": "k/c", "detail": "k/d"}) + assert repo.claim_image_for_fetch(db, img) is False + + +def test_mark_image_outcomes_and_run_progress(db, seed_storefront): + sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images") + p = _seed_product(db, sf, "lamp") + ok = _seed_image(db, p, "https://m/a.png", run_id, status="pending") + bad = _seed_image(db, p, "https://m/b.png", run_id, status="pending") + miss = _seed_image(db, p, "https://m/c.png", run_id, status="pending") + repo.mark_image_fetched(db, ok, {"original": "o", "thumb": "t", "card": "c", "detail": "d"}) + repo.mark_image_rejected(db, bad, "rejected_low_res", "below the resolution bar") + repo.mark_image_failed(db, miss, "host unreachable") + counts = repo.run_image_counts(db, run_id) + assert counts == {"fetched": 1, "rejected": 1, "failed": 1, "pending": 0, "total": 3} + + +def test_incomplete_runs_lists_fetching_images_runs(db, seed_storefront): + sf = seed_storefront(db) + rid = _seed_run(db, sf, status="fetching_images") + _seed_run(db, sf, status="complete") + assert rid in [r["id"] for r in repo.incomplete_runs(db)] +``` + +(Define the small `_seed_*` helpers inline in the test using the existing insert primitives; `db`/`seed_storefront` mirror whatever conftest already provides — check `backend/tests/conftest.py` and reuse its names.) + +- [ ] **Step 2: Run to verify failure** — Run: `./.venv/bin/python -m pytest backend/tests/test_products_repo_images.py -q`. Expected: FAIL (helpers missing). + +- [ ] **Step 3: Implement the repo helpers** + +Add to `repo.py`: + +```python +# --------------------------------------------------------------------------- +# Image-fetch phase (SD-0002 §6.5.4) — called by domains/products/imagefetch.py +# --------------------------------------------------------------------------- + +def pending_images_for_run(conn: psycopg.Connection, run_id: int) -> list[dict]: + """The run's images still awaiting fetch, with the storefront for key layout.""" + rows = conn.execute( + "SELECT i.id, i.source_url, p.storefront_id, i.product_id" + " FROM product_image i JOIN product p ON p.id = i.product_id" + " WHERE i.import_run_id = %s AND i.status = 'pending'" + " ORDER BY i.id", + (run_id,), + ).fetchall() + return [ + {"id": r[0], "source_url": r[1], "storefront_id": r[2], "product_id": r[3]} + for r in rows + ] + + +def claim_image_for_fetch(conn: psycopg.Connection, image_id: int) -> bool: + """Atomically claim a pending image (idempotent restart guard). Returns True + iff this caller won the claim (it was still 'pending').""" + row = conn.execute( + "UPDATE product_image SET status = 'pending' WHERE id = %s AND status = 'pending'" + " RETURNING id", + (image_id,), + ).fetchone() + return row is not None + + +def mark_image_fetched(conn: psycopg.Connection, image_id: int, keys: dict[str, str]) -> None: + conn.execute( + "UPDATE product_image SET status = 'fetched', failure_reason = NULL," + " key_original = %s, key_thumb = %s, key_card = %s, key_detail = %s," + " fetched_at = now() WHERE id = %s", + (keys["original"], keys["thumb"], keys["card"], keys["detail"], image_id), + ) + + +def mark_image_rejected(conn: psycopg.Connection, image_id: int, status: str, reason: str) -> None: + conn.execute( + "UPDATE product_image SET status = %s, failure_reason = %s, fetched_at = now()" + " WHERE id = %s", + (status, reason, image_id), + ) + + +def mark_image_failed(conn: psycopg.Connection, image_id: int, reason: str) -> None: + conn.execute( + "UPDATE product_image SET status = 'failed', failure_reason = %s, fetched_at = now()" + " WHERE id = %s", + (reason, image_id), + ) + + +def run_image_counts(conn: psycopg.Connection, run_id: int) -> dict: + row = conn.execute( + "SELECT" + " count(*) FILTER (WHERE status = 'fetched')," + " count(*) FILTER (WHERE status IN ('rejected_low_res', 'rejected_not_image'))," + " count(*) FILTER (WHERE status = 'failed')," + " count(*) FILTER (WHERE status = 'pending')," + " count(*)" + " FROM product_image WHERE import_run_id = %s", + (run_id,), + ).fetchone() + return {"fetched": row[0], "rejected": row[1], "failed": row[2], + "pending": row[3], "total": row[4]} + + +def set_run_status(conn: psycopg.Connection, run_id: int, status: str) -> None: + conn.execute( + "UPDATE import_run SET status = %s," + " completed_at = CASE WHEN %s THEN now() ELSE completed_at END" + " WHERE id = %s", + (status, status in _TERMINAL_RUN_STATUSES, run_id), + ) + + +def incomplete_runs(conn: psycopg.Connection) -> list[dict]: + """Runs still in the image phase — the startup recovery scan input (§6.9).""" + rows = conn.execute( + "SELECT id, storefront_id FROM import_run WHERE status = 'fetching_images'" + " ORDER BY id", + ).fetchall() + return [{"id": r[0], "storefront_id": r[1]} for r in rows] + + +def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]: + """Problem images for the run-detail outcomes table (only non-fetched).""" + rows = conn.execute( + "SELECT p.handle, i.source_url, i.status, i.failure_reason," + " v.option1_value, v.option2_value, v.option3_value" + " FROM product_image i JOIN product p ON p.id = i.product_id" + " LEFT JOIN variant v ON v.image_id = i.id" + " WHERE i.import_run_id = %s" + " AND i.status IN ('rejected_low_res', 'rejected_not_image', 'failed')" + " ORDER BY p.handle, i.position, i.id", + (run_id,), + ).fetchall() + return [ + {"handle": r[0], "url": r[1], "outcome": r[2], "reason": r[3], + "variant": " / ".join(x for x in (r[4], r[5], r[6]) if x) or None} + for r in rows + ] +``` + +- [ ] **Step 4: Wire image progress/outcomes/counts into `get_run` and `list_runs`** + +Replace the SLICE-5 stub in `get_run`: + +```python + counts = run_image_counts(conn, run_id) + run["image_progress"] = {"done": counts["total"] - counts["pending"], "total": counts["total"]} + run["image_counts"] = {"fetched": counts["fetched"], "rejected": counts["rejected"], + "failed": counts["failed"]} + run["image_outcomes"] = run_image_outcomes(conn, run_id) +``` + +For `list_runs`, add per-run image counts so the history table can show "✓ · ✗". Simplest: in `_run_dict` leave as-is, and have `list_runs` attach `image_counts` per run via one grouped query: + +```python +def list_runs(conn, storefront_id, limit, offset): + rows = conn.execute(_RUN_SELECT + " WHERE r.storefront_id = %s" + " ORDER BY r.created_at DESC, r.id DESC LIMIT %s OFFSET %s", + (storefront_id, limit, offset)).fetchall() + runs = [_run_dict(row) for row in rows] + if not runs: + return runs + ids = [r["id"] for r in runs] + by_run = {rid: {"fetched": 0, "rejected": 0, "failed": 0, "total": 0} for rid in ids} + for rid, fetched, rejected, failed, total in conn.execute( + "SELECT import_run_id," + " count(*) FILTER (WHERE status='fetched')," + " count(*) FILTER (WHERE status IN ('rejected_low_res','rejected_not_image'))," + " count(*) FILTER (WHERE status='failed'), count(*)" + " FROM product_image WHERE import_run_id = ANY(%s) GROUP BY import_run_id", + (ids,), + ): + by_run[rid] = {"fetched": fetched, "rejected": rejected, "failed": failed, "total": total} + for r in runs: + r["image_counts"] = by_run[r["id"]] + return runs +``` + +- [ ] **Step 5: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_repo_images.py backend/tests/test_products_service.py -q` +Expected: PASS. Note: `test_products_service.py`'s existing assertion that `image_progress` is `{done:0,total:0}` for a no-image run still holds (total 0). + +- [ ] **Step 6: Commit** + +```bash +git add backend/app/domains/products/repo.py backend/tests/test_products_repo_images.py +git commit -m "feat(products): image-phase repo helpers + run progress/outcomes/counts (SD-0002 §6.5.4)" +``` + +--- + +## Task 7: the image-fetch phase + SSRF guard + recovery + +**Files:** +- Create: `backend/app/domains/products/imagefetch.py` +- Create: `backend/tests/test_products_imagefetch.py` +- Modify: `backend/app/domains/products/__init__.py` + +- [ ] **Step 1: Write the failing tests (SSRF guard unit; fetch against a local fixture server; resume)** + +`backend/tests/test_products_imagefetch.py`: + +```python +"""image-fetch phase: SSRF guard, fetch+process+store, run completion, resume (§6.5.4/§6.9).""" +import io +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest +from PIL import Image + +from app.domains.products import imagefetch +from app.platform import objectstore + + +def _png(w, h): + buf = io.BytesIO(); Image.new("RGB", (w, h), (1, 2, 3)).save(buf, "PNG"); return buf.getvalue() + + +class _Host(BaseHTTPRequestHandler): + GOOD = _png(800, 800) + TINY = _png(64, 64) + def log_message(self, *a): pass + def do_GET(self): + if self.path == "/good.png": + self.send_response(200); self.send_header("Content-Type", "image/png") + self.end_headers(); self.wfile.write(self.GOOD) + elif self.path == "/tiny.png": + self.send_response(200); self.send_header("Content-Type", "image/png") + self.end_headers(); self.wfile.write(self.TINY) + else: + self.send_response(404); self.end_headers() + + +@pytest.fixture +def host(): + srv = HTTPServer(("127.0.0.1", 0), _Host) + t = threading.Thread(target=srv.serve_forever, daemon=True); t.start() + yield f"http://127.0.0.1:{srv.server_address[1]}" + srv.shutdown() + + +def test_ssrf_guard_rejects_private_when_disallowed(): + with pytest.raises(imagefetch.FetchBlocked): + imagefetch.fetch_bytes("http://127.0.0.1:1/x.png", allow_private=False) + with pytest.raises(imagefetch.FetchBlocked): + imagefetch.fetch_bytes("http://169.254.169.254/latest/meta-data", allow_private=False) + with pytest.raises(imagefetch.FetchBlocked): + imagefetch.fetch_bytes("ftp://example.com/x", allow_private=False) + + +def test_ssrf_guard_allows_private_when_enabled(host): + data, ctype = imagefetch.fetch_bytes(f"{host}/good.png", allow_private=True) + assert ctype.startswith("image/") + assert len(data) > 0 + + +def test_run_phase_classifies_each_image(db, seed_storefront, host, tmp_path): + store = objectstore.LocalObjectStore(str(tmp_path)) + sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images") + p = _seed_product(db, sf, "lamp") + ok = _seed_image(db, p, f"{host}/good.png", run_id) + tiny = _seed_image(db, p, f"{host}/tiny.png", run_id) + miss = _seed_image(db, p, f"{host}/missing.png", run_id) + imagefetch.run_image_phase(_pool_for(db), store, run_id, allow_private=True) + counts = repo.run_image_counts(db, run_id) + assert counts["fetched"] == 1 and counts["rejected"] == 1 and counts["failed"] == 1 + # fetched image got four objects stored + keys = _image_keys(db, ok) + for k in ("key_original", "key_thumb", "key_card", "key_detail"): + assert keys[k] and store.get(keys[k]) + # run reached a terminal status + assert _run_status(db, run_id) == "complete_with_problems" + + +def test_resume_after_kill_completes_remaining(db, seed_storefront, host, tmp_path): + store = objectstore.LocalObjectStore(str(tmp_path)) + sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images") + p = _seed_product(db, sf, "lamp") + a = _seed_image(db, p, f"{host}/good.png", run_id) + # simulate a crash: one image already fetched, run still fetching_images + repo.mark_image_fetched(db, a, {"original": "o", "thumb": "t", "card": "c", "detail": "d"}) + b = _seed_image(db, p, f"{host}/good.png?2", run_id) # still pending + resumed = imagefetch.recover_incomplete_runs(_pool_for(db), store, allow_private=True) + assert run_id in resumed + assert repo.run_image_counts(db, run_id)["pending"] == 0 + assert _run_status(db, run_id) == "complete" +``` + +(`_pool_for(db)` returns a minimal object exposing a `connection()` context manager yielding `db` — or use the real test pool fixture if conftest exposes one. `_seed_*`, `_image_keys`, `_run_status` are inline helpers over raw SQL.) + +- [ ] **Step 2: Run to verify failure** — `./.venv/bin/python -m pytest backend/tests/test_products_imagefetch.py -q`. Expected: FAIL (module missing). + +- [ ] **Step 3: Implement `imagefetch.py`** + +```python +"""The post-commit image-fetch phase (SD-0002 §6.5.4, §6.9). + +In-process, bounded-concurrency, resumable. For a run's pending images: fetch +each URL behind an SSRF guard with INV-18 bounds, run platform/images, store +renditions via the objectstore, and move the image pending -> fetched | +rejected_* | failed. Per-image idempotent (claim guard), so the phase can die +and resume. Lives in the domains layer (imports repo + platform); main.py +schedules it post-confirm and runs recovery at startup. +""" +from __future__ import annotations + +import ipaddress +import socket +import time +from concurrent.futures import ThreadPoolExecutor +from urllib.parse import urlsplit + +import httpx + +from app.platform import images as images_mod +from app.platform import objectstore as objectstore_mod +from app.platform import telemetry + +from . import repo + +MAX_FETCH_BYTES = 20 * 1024 * 1024 # INV-18 +FETCH_TIMEOUT_S = 30 # INV-18 +FETCH_CONCURRENCY = 4 # §6.6 +_CONTENT_PREFIX = "image/" + + +class FetchBlocked(Exception): + """SSRF guard or bounds refusal — maps to image status 'failed' with reason.""" + + +def _guard_host(url: str, allow_private: bool) -> None: + parts = urlsplit(url) + if parts.scheme not in {"http", "https"}: + raise FetchBlocked(f"unsupported scheme: {parts.scheme!r}") + host = parts.hostname + if not host: + raise FetchBlocked("no host") + if allow_private: + return + # Resolve and reject private / loopback / link-local / reserved / multicast. + try: + infos = socket.getaddrinfo(host, parts.port or (443 if parts.scheme == "https" else 80)) + except socket.gaierror as exc: + raise FetchBlocked(f"dns failure: {exc}") from exc + for info in infos: + ip = ipaddress.ip_address(info[4][0]) + if (ip.is_private or ip.is_loopback or ip.is_link_local + or ip.is_reserved or ip.is_multicast or ip.is_unspecified): + raise FetchBlocked(f"non-public address: {ip}") + + +def fetch_bytes(url: str, allow_private: bool) -> tuple[bytes, str]: + """SSRF-guarded, bounded GET. Returns (bytes, content_type). Raises FetchBlocked.""" + _guard_host(url, allow_private) + # No redirects across the guard (§6.6): a 3xx is treated as a failure. + with httpx.Client(follow_redirects=False, timeout=FETCH_TIMEOUT_S) as client: + with client.stream("GET", url) as resp: + if resp.status_code != 200: + raise FetchBlocked(f"http {resp.status_code}") + ctype = resp.headers.get("content-type", "").split(";")[0].strip().lower() + if not ctype.startswith(_CONTENT_PREFIX): + raise FetchBlocked(f"not an image content-type: {ctype!r}") + chunks: list[bytes] = [] + total = 0 + for chunk in resp.iter_bytes(): + total += len(chunk) + if total > MAX_FETCH_BYTES: + raise FetchBlocked("oversize") + chunks.append(chunk) + return b"".join(chunks), ctype + + +def _key(storefront_id: int, image_id: int, name: str) -> str: + return f"storefronts/{storefront_id}/product-images/{image_id}/{name}" + + +def _process_one(pool, store, image: dict, allow_private: bool) -> None: + """One image, its own short transaction(s). Idempotent via the claim guard.""" + with pool.connection() as conn: + claimed = repo.claim_image_for_fetch(conn, image["id"]) + conn.commit() + if not claimed: + return + image_id, sf = image["id"], image["storefront_id"] + try: + data, ctype = fetch_bytes(image["source_url"], allow_private) + except (FetchBlocked, httpx.HTTPError) as exc: + with pool.connection() as conn: + repo.mark_image_failed(conn, image_id, str(exc)[:200]); conn.commit() + return + result = images_mod.process(data) + if isinstance(result, images_mod.Rejected): + reason = ("below the resolution bar" if result.reason == "rejected_low_res" + else "not a supported image") + with pool.connection() as conn: + repo.mark_image_rejected(conn, image_id, result.reason, reason); conn.commit() + return + keys = {"original": _key(sf, image_id, "original"), + "thumb": _key(sf, image_id, "thumb.webp"), + "card": _key(sf, image_id, "card.webp"), + "detail": _key(sf, image_id, "detail.webp")} + store.put(keys["original"], data, ctype) + store.put(keys["thumb"], result.renditions["thumb"], "image/webp") + store.put(keys["card"], result.renditions["card"], "image/webp") + store.put(keys["detail"], result.renditions["detail"], "image/webp") + with pool.connection() as conn: + repo.mark_image_fetched(conn, image_id, keys); conn.commit() + + +def run_image_phase(pool, store, run_id: int, allow_private: bool) -> None: + """Fetch all pending images of one run, then move the run to a terminal status.""" + started = time.monotonic() + with pool.connection() as conn: + pending = repo.pending_images_for_run(conn, run_id) + if pending: + with ThreadPoolExecutor(max_workers=FETCH_CONCURRENCY) as pool_x: + list(pool_x.map(lambda img: _process_one(pool, store, img, allow_private), pending)) + with pool.connection() as conn: + counts = repo.run_image_counts(conn, run_id) + status = ("complete_with_problems" + if counts["rejected"] + counts["failed"] > 0 else "complete") + repo.set_run_status(conn, run_id, status) + conn.commit() + telemetry.emit("image_phase_completed", run_id=run_id, fetched=counts["fetched"], + rejected=counts["rejected"], failed=counts["failed"], + duration_ms=int((time.monotonic() - started) * 1000)) + + +def recover_incomplete_runs(pool, store, allow_private: bool) -> list[int]: + """Startup scan (§6.9): resume any run stuck in fetching_images. Returns the ids.""" + with pool.connection() as conn: + runs = repo.incomplete_runs(conn) + resumed: list[int] = [] + for run in runs: + with pool.connection() as conn: + pending = len(repo.pending_images_for_run(conn, run["id"])) + telemetry.emit("image_phase_recovered", run_id=run["id"], pending_resumed=pending) + run_image_phase(pool, store, run["id"], allow_private) + resumed.append(run["id"]) + return resumed +``` + +- [ ] **Step 4: Export the entrypoints** — add to `domains/products/__init__.py` `__all__` and imports: `run_image_phase`, `recover_incomplete_runs` (re-export from `.imagefetch`). Keep the existing exports. + +- [ ] **Step 5: Run to verify pass** + +Run: `./.venv/bin/python -m pytest backend/tests/test_products_imagefetch.py -q` +Expected: PASS (SSRF unit + fixture-host classify + resume). + +- [ ] **Step 6: Verify layering** — `cd backend && ../.venv/bin/lint-imports`. Expected: kept (imagefetch is in domains, imports platform only). + +- [ ] **Step 7: Commit** + +```bash +git add backend/app/domains/products/imagefetch.py backend/app/domains/products/__init__.py backend/tests/test_products_imagefetch.py +git commit -m "feat(products): image-fetch phase — SSRF guard, bounds, renditions, resume (SD-0002 §6.5.4, §6.9, INV-18)" +``` + +--- + +## Task 8: confirm sets `fetching_images`; lifespan wiring + post-commit scheduling + +**Files:** +- Modify: `backend/app/domains/products/service.py` (`confirm_draft` run status) +- Modify: `backend/app/main.py` (lifespan, scheduler, recovery) +- Modify: `backend/tests/test_products_service.py` + +- [ ] **Step 1: Write the failing test (confirm marks the run fetching_images when it has pending images; complete otherwise)** + +Add to `backend/tests/test_products_service.py`: + +```python +def test_confirm_marks_run_fetching_images_when_images_present(db, seed_storefront): + # Import a product whose CSV references an image URL → run should be + # 'fetching_images' (the post-commit task is scheduled by the BFF, not here). + ... # build a draft from a CSV with an Image Src, confirm, assert run status + run = service.get_run(db, sf, run_id) + assert run["status"] == "fetching_images" + assert run["image_progress"]["total"] >= 1 + + +def test_confirm_marks_run_complete_when_no_images(db, seed_storefront): + ... # CSV with no Image Src → run status 'complete' immediately + assert service.get_run(db, sf, run_id)["status"] == "complete" +``` + +- [ ] **Step 2: Run to verify failure** — Expected: FAIL (confirm always writes 'complete'). + +- [ ] **Step 3: Make `confirm_draft` choose the status by pending-image presence** + +In `service.py` `confirm_draft`, after applying plans and before/at commit, decide the run status. Since images are inserted as `pending` during apply, count them on the just-applied run; insert the run as `applying` first, then set the terminal/transitional status at the end of the transaction: + +```python + run_id = repo.insert_run( + conn, storefront_id, account_id, row["file_name"], row["dialect"], + added=summary_counts["adds"], updated=summary_counts["updates"], + errored=len(error_rows), status="applying", + ) + for plan in diff_result.plan: + _apply_product_plan(conn, storefront_id, plan, run_id) + repo.insert_run_errors(conn, run_id, error_rows) + pending = repo.run_image_counts(conn, run_id)["pending"] + repo.set_run_status(conn, run_id, "fetching_images" if pending else "complete") + repo.delete_draft(conn, storefront_id, draft_id) + conn.commit() +``` + +`insert_run` with `status="applying"` writes no `completed_at` (it's not terminal); `set_run_status("complete")` then stamps it. TEL-2 (`import_run_completed`) still fires after commit as today — it marks the *catalog* apply, independent of the image phase. + +- [ ] **Step 4: Wire lifespan, recovery, scheduler in `main.py`** + +```python +from app.platform import objectstore as objectstore_mod +from app.domains import products +from concurrent.futures import ThreadPoolExecutor + +@asynccontextmanager +async def lifespan(app: FastAPI): + app.state.pool = db.open_pool(dsn) + with app.state.pool.connection() as conn: + db.migrate(conn) + app.state.mailer = mailer_mod.build_mailer(config.mailer_kind()) + app.state.objectstore = objectstore_mod.build_objectstore(config.objectstore_kind()) + app.state.image_runner = ThreadPoolExecutor(max_workers=2, thread_name_prefix="img-run") + allow_private = config.image_fetch_allow_private() + app.state.image_allow_private = allow_private + # §6.9 startup recovery — resume runs stuck mid image-fetch (off the request path) + app.state.image_runner.submit( + products.recover_incomplete_runs, app.state.pool, app.state.objectstore, allow_private + ) + try: + yield + finally: + app.state.image_runner.shutdown(wait=False) + app.state.pool.close() +``` + +In the confirm endpoint, after `confirm_draft` returns `run_id`, schedule the phase off the request thread (only if the run has work — cheap to always submit; the phase no-ops when there are no pending images): + +```python + run_id = products.confirm_draft(conn, sf.id, account.id, draft_id) + request.app.state.image_runner.submit( + products.run_image_phase, request.app.state.pool, + request.app.state.objectstore, run_id, request.app.state.image_allow_private, + ) + return JSONResponse({"run_id": run_id}, status_code=201) +``` + +(Add `request: Request` to the confirm handler signature if not present.) + +- [ ] **Step 5: Run to verify pass** — `./.venv/bin/python -m pytest backend/tests/test_products_service.py -q`. Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add backend/app/domains/products/service.py backend/app/main.py backend/tests/test_products_service.py +git commit -m "feat(products): confirm → fetching_images + post-commit fetch scheduling + startup recovery (SD-0002 §6.5.3/§6.9)" +``` + +--- + +## Task 9: app-served image route (INV-16) + +**Files:** +- Modify: `backend/app/main.py` +- Create: `backend/tests/test_products_image_serving.py` + +- [ ] **Step 1: Write the failing endpoint tests** + +`backend/tests/test_products_image_serving.py` (use the existing TestClient `@contextmanager` app fixture; the test app uses the local objectstore; seed a fetched image with stored bytes and a second storefront for isolation): + +```python +"""GET /api/products/images/{id}/{rendition} — authorize, stream, immutable cache (§6.4, INV-14/16).""" + +def test_serves_fetched_rendition_with_immutable_cache(client, signed_in_with_image): + image_id = signed_in_with_image # a fetched image owned by the session's storefront + resp = client.get(f"/api/products/images/{image_id}/detail") + assert resp.status_code == 200 + assert resp.headers["content-type"] == "image/webp" + assert "immutable" in resp.headers.get("cache-control", "") + assert resp.content # bytes streamed + + +def test_not_fetched_returns_409(client, signed_in_with_pending_image): + image_id = signed_in_with_pending_image + resp = client.get(f"/api/products/images/{image_id}/detail") + assert resp.status_code == 409 # not_fetched + + +def test_other_storefronts_image_is_404(client, image_of_other_storefront): + resp = client.get(f"/api/products/images/{image_of_other_storefront}/detail") + assert resp.status_code == 404 + + +def test_unauthenticated_is_401(client_no_session, any_image_id): + resp = client_no_session.get(f"/api/products/images/{any_image_id}/detail") + assert resp.status_code == 401 + + +def test_bad_rendition_is_422(client, signed_in_with_image): + resp = client.get(f"/api/products/images/{signed_in_with_image}/huge") + assert resp.status_code == 422 # path pattern rejects unknown rendition +``` + +- [ ] **Step 2: Run to verify failure** — route missing → 404/405 mismatches. Expected: FAIL. + +- [ ] **Step 3: Implement the route + repo lookup** + +Add a repo helper `image_storage(conn, storefront_id, image_id)` returning the storage key + status for a rendition, storefront-scoped: + +```python +def image_for_serving(conn, storefront_id, image_id): + 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]} +``` + +In `main.py`: + +```python +from fastapi import Response +from app.platform import objectstore as objectstore_mod + +_RENDITION_CT = {"original": None, "thumb": "image/webp", "card": "image/webp", "detail": "image/webp"} + +@app.get("/api/products/images/{image_id}/{rendition}") +def serve_product_image( + image_id: int, + rendition: str = Path(..., pattern="^(original|thumb|card|detail)$"), + request: Request = None, + conn: psycopg.Connection = Depends(get_conn), + sess: dict | None = Depends(get_session), +): + 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 = request.app.state.objectstore.get(key) + except objectstore_mod.ObjectNotFound: + return _error(404, "not_found", "No such image.") + media_type = _RENDITION_CT[rendition] or "application/octet-stream" + return Response(content=data, media_type=media_type, + headers={"Cache-Control": "public, max-age=31536000, immutable"}) +``` + +Export `image_for_serving` from `domains/products/__init__.py`. (FastAPI returns `422` automatically for a path that fails the `pattern` — covers the bad-rendition test.) + +- [ ] **Step 4: Run to verify pass** — `./.venv/bin/python -m pytest backend/tests/test_products_image_serving.py -q`. Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add backend/app/main.py backend/app/domains/products/repo.py backend/app/domains/products/__init__.py backend/tests/test_products_image_serving.py +git commit -m "feat(products): app-served image route — storefront-authorized, immutable cache (SD-0002 §6.4, INV-16)" +``` + +--- + +## Task 10: frontend — run-detail images section, notice band, history column + +**Files:** +- Modify: `frontend/src/productsApi.ts` +- Modify: `frontend/src/screens/products/RunDetail.tsx` +- Modify: `frontend/src/screens/products/ProductsPage.tsx` +- Create: `frontend/src/screens/products/runImages.ts` + `frontend/src/screens/products/runImages.test.ts` + +- [ ] **Step 1: Type the payloads + pure helpers (vitest first)** + +In `productsApi.ts`: + +```typescript +export interface ImageOutcome { + handle: string; url: string; outcome: string; reason: string | null; variant: string | null; +} +export interface ImageCounts { fetched: number; rejected: number; failed: number; } +export interface RunDetail extends RunSummary { + errors: RowErrorDetail[]; + image_progress: { done: number; total: number }; + image_counts: ImageCounts; + image_outcomes: ImageOutcome[]; +} +export interface RunSummary { + /* ...existing... */ image_counts?: ImageCounts; +} +``` + +`runImages.ts`: + +```typescript +import type { ImageCounts } from "../../productsApi"; + +export function imageProgressLabel(p: { done: number; total: number }): string { + return `Fetching images: ${p.done.toLocaleString()} of ${p.total.toLocaleString()}`; +} + +export function imageOutcomeSummary(c: ImageCounts): string { + return `${c.fetched.toLocaleString()} fetched · ${c.rejected.toLocaleString()} rejected · ${c.failed.toLocaleString()} failed`; +} + +export function historyImageCell(c: ImageCounts | undefined): string { + if (!c || c.fetched + c.rejected + c.failed === 0) return "—"; + const bad = c.rejected + c.failed; + return bad === 0 ? `${c.fetched.toLocaleString()} ✓` : `${c.fetched.toLocaleString()} ✓ · ${bad.toLocaleString()} ✗`; +} + +export function isRunTerminal(status: string): boolean { + return status === "complete" || status === "complete_with_problems"; +} +``` + +`runImages.test.ts`: + +```typescript +import { describe, expect, it } from "vitest"; +import { historyImageCell, imageOutcomeSummary, imageProgressLabel, isRunTerminal } from "./runImages"; + +describe("runImages helpers", () => { + it("progress label", () => { + expect(imageProgressLabel({ done: 312, total: 4950 })).toBe("Fetching images: 312 of 4,950"); + }); + it("outcome summary", () => { + expect(imageOutcomeSummary({ fetched: 4938, rejected: 9, failed: 3 })) + .toBe("4,938 fetched · 9 rejected · 3 failed"); + }); + it("history cell collapses clean / shows problems / dashes empty", () => { + expect(historyImageCell({ fetched: 10, rejected: 0, failed: 0 })).toBe("10 ✓"); + expect(historyImageCell({ fetched: 8, rejected: 1, failed: 1 })).toBe("8 ✓ · 2 ✗"); + expect(historyImageCell(undefined)).toBe("—"); + }); + it("terminal status", () => { + expect(isRunTerminal("fetching_images")).toBe(false); + expect(isRunTerminal("complete_with_problems")).toBe(true); + }); +}); +``` + +Run: `cd frontend && npm test -- runImages` → FAIL (helpers missing), then PASS after creating `runImages.ts`. + +- [ ] **Step 2: RunDetail — images section + polling** + +In `RunDetail.tsx`: remove the `// NO images section this slice (SLICE-7)` comment; add a `useEffect` that re-fetches the run on an interval (e.g. 2 s) while `!isRunTerminal(run.status)`, clearing on terminal/unmount. Render: +- while `!isRunTerminal`: `

{imageProgressLabel(run.image_progress)}

` (status line "Fetching images…"). +- when terminal and `image_progress.total > 0`: `

{imageOutcomeSummary(run.image_counts)}

` and, if `image_outcomes.length`, an outcomes table: columns Handle · Variant · Image URL · Outcome · What to do. Map `outcome` to label ("rejected: below the resolution bar / not an image", "failed: unreachable") and a fixed "Correct the URL and re-import." action. All-clean (`image_outcomes.length === 0`) collapses to one line "All images fetched." + +- [ ] **Step 3: ProductsPage — notice band + history image column** + +In `ProductsPage.tsx`: +- Notice band (after header, before content): when `summary.image_problem_count > 0`, render `
{summary.image_problem_count} products have image problems
` (aria-live). Uses the existing `getProductsSummary` payload (already returns `image_problem_count`, `latest_run_id`). +- History table: add an "Images" column header and a cell `{historyImageCell(r.image_counts)}` between Errors and Status. + +- [ ] **Step 4: Build + unit gate** + +Run: `cd frontend && npm run build && npm test` +Expected: typecheck + build clean; vitest green (incl. runImages). + +- [ ] **Step 5: Commit** + +```bash +git add frontend/src/productsApi.ts frontend/src/screens/products/RunDetail.tsx frontend/src/screens/products/ProductsPage.tsx frontend/src/screens/products/runImages.ts frontend/src/screens/products/runImages.test.ts +git commit -m "feat(products-ui): run-detail images section + progress poll, image-problems notice band, history images column (SD-0002 §5.2/§5.5)" +``` + +--- + +## Task 11: E2E — `e2e_image_outcomes` against a fixture image host + +**Files:** +- Create: `e2e/image-host.mjs`, `e2e/fixtures/images/good.png`, `e2e/fixtures/images/tiny.png`, `e2e/fixtures/with-images.csv` +- Modify: `e2e/serve.sh`, `e2e/helpers.ts`, create `e2e/tests/image-outcomes.spec.ts` + +- [ ] **Step 1: Create the fixture images + CSV** + +Generate the PNGs (one ≥500px good, one tiny) — use Python/Pillow once, commit the bytes: + +```bash +./.venv/bin/python - <<'PY' +from PIL import Image +Image.new("RGB",(800,800),(40,120,200)).save("e2e/fixtures/images/good.png") +Image.new("RGB",(64,64),(200,40,40)).save("e2e/fixtures/images/tiny.png") +PY +``` + +`e2e/fixtures/with-images.csv` (canonical; three image URLs against the fixture host on :8799 — good / tiny(low-res) / missing(404)): + +```csv +Handle,Title,Image Src +good-lamp,Good Lamp,http://127.0.0.1:8799/good.png +tiny-lamp,Tiny Lamp,http://127.0.0.1:8799/tiny.png +gone-lamp,Gone Lamp,http://127.0.0.1:8799/missing.png +``` + +- [ ] **Step 2: Fixture image host** + +`e2e/image-host.mjs` — a tiny static server on :8799 serving `fixtures/images/*` and 404 for anything else: + +```javascript +import { createServer } from "node:http"; +import { readFile } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; + +const dir = join(dirname(fileURLToPath(import.meta.url)), "fixtures", "images"); +createServer(async (req, res) => { + const name = (req.url || "").replace(/^\//, "").split("?")[0]; + if (name === "good.png" || name === "tiny.png") { + try { + const buf = await readFile(join(dir, name)); + res.writeHead(200, { "Content-Type": "image/png" }); + res.end(buf); + return; + } catch { /* fallthrough */ } + } + res.writeHead(404); + res.end(); +}).listen(8799, "127.0.0.1"); +``` + +- [ ] **Step 3: serve.sh — start the host + set image env** + +In `e2e/serve.sh`, before launching uvicorn, add the objectstore + fetch env and start the host in the background: + +```bash +export ECOMM_OBJECTSTORE_KIND=local +export ECOMM_OBJECTSTORE_DIR="$repo_root/e2e/.objectstore" +export ECOMM_IMAGE_FETCH_ALLOW_PRIVATE=1 +rm -rf "$repo_root/e2e/.objectstore" +node "$repo_root/e2e/image-host.mjs" & +``` + +(Add `.objectstore` to `e2e/.gitignore`.) The host is a daemon of the serve process; Playwright tears the whole webServer down between runs. + +- [ ] **Step 4: helper + spec** + +`e2e/helpers.ts` — add `importWithImages(page)` (upload `with-images.csv`, confirm, land on run detail) mirroring `importGoodCsv`. `e2e/tests/image-outcomes.spec.ts`: + +```typescript +import { expect, test } from "@playwright/test"; +import { gotoProducts, importWithImages, signUpWithStorefront } from "../helpers"; + +test("e2e_image_outcomes: per-image outcomes, placeholder, notice band", async ({ page }) => { + await signUpWithStorefront(page); + await importWithImages(page); + // Run detail polls fetching_images → terminal. Wait for the outcome summary. + await expect(page.getByText(/fetched ·/)).toBeVisible({ timeout: 30_000 }); + // one good (fetched), one tiny (rejected), one missing (failed) + await expect(page.getByText("1 fetched · 1 rejected · 1 failed")).toBeVisible(); + // problems listed in the outcomes table + await expect(page.getByText("tiny-lamp")).toBeVisible(); + await expect(page.getByText("gone-lamp")).toBeVisible(); + // Products page surfaces the aggregate notice + await gotoProducts(page); + await expect(page.getByText(/products have image problems/)).toBeVisible(); +}); +``` + +- [ ] **Step 5: Run the full E2E suite** + +Run: `bash scripts/e2e.sh` +Expected: 7/7 green (the 6 existing + `e2e_image_outcomes`). If the SPA poll timing flakes, raise the `expect(...).toBeVisible` timeout; the fetch of two local images completes in <2 s. + +- [ ] **Step 6: Commit** + +```bash +git add e2e/ +git commit -m "test(e2e): e2e_image_outcomes — fixture image host, per-image outcomes + notice band (SD-0002 §6.8)" +``` + +--- + +## Task 12: flotilla/launch-app `provision-bucket` gesture (engineering repo) + +**Files (in `~/git/wiggleverse.org/wiggleverse/engineering`):** +- Create: `launch-app/skills/provision-bucket/SKILL.md` +- Create: `launch-app/skills/provision-bucket/assets/provision-bucket.sh` +- Modify: `launch-app/SPEC.md` (add §6.6b) + +This is built by this session (dev work), mirroring `provision-datastore`; the operator *runs* it at the PPE deploy (Task 14). Work on a branch in the engineering repo, PR, merge. + +- [ ] **Step 1: Branch the engineering repo** + +```bash +cd ~/git/wiggleverse.org/wiggleverse/engineering +git fetch origin --quiet && git switch -c provision-bucket-gesture origin/main +``` + +- [ ] **Step 2: Write `assets/provision-bucket.sh`** (idempotent; probes before create; selects the dedicated gcloud config; never touches the shared active pointer) + +```bash +#!/usr/bin/env bash +# launch-app §6.6b — provision the per-environment private media bucket +# (SD-0002 §6.3.1). Idempotent. Reads params from read-deployment-params.py. +set -euo pipefail + +: "${NAME:?}"; : "${PROJECT:?}"; : "${GCLOUD_CONFIG:?}" +REGION="${REGION:-us-central1}" +ENVNAME="${ENVNAME:-ppe}" +BUCKET="${BUCKET:-${PROJECT}-${ENVNAME}-media}" +VM_SA="${VM_SA:-$(CLOUDSDK_ACTIVE_CONFIG_NAME=$GCLOUD_CONFIG gcloud compute project-info describe \ + --project "$PROJECT" --format='value(defaultServiceAccount)')}" + +run() { CLOUDSDK_ACTIVE_CONFIG_NAME="$GCLOUD_CONFIG" "$@"; } + +echo "==> enabling storage API" +run gcloud services enable storage.googleapis.com --project "$PROJECT" + +if run gcloud storage buckets describe "gs://$BUCKET" --project "$PROJECT" >/dev/null 2>&1; then + echo "==> bucket gs://$BUCKET exists" +else + echo "==> creating gs://$BUCKET (Standard, uniform access, no public, region $REGION)" + run gcloud storage buckets create "gs://$BUCKET" \ + --project "$PROJECT" --location "$REGION" \ + --uniform-bucket-level-access --public-access-prevention --default-storage-class STANDARD +fi + +echo "==> lifecycle: delete import-drafts/ objects after 1 day (§6.3.1 safety net)" +tmp=$(mktemp) +cat > "$tmp" <<'JSON' +{"rule":[{"action":{"type":"Delete"},"condition":{"age":1,"matchesPrefix":["import-drafts/"]}}]} +JSON +run gcloud storage buckets update "gs://$BUCKET" --project "$PROJECT" --lifecycle-file="$tmp" +rm -f "$tmp" + +echo "==> granting objectAdmin on the bucket to the VM service account" +run gcloud storage buckets add-iam-policy-binding "gs://$BUCKET" --project "$PROJECT" \ + --member="serviceAccount:${VM_SA}" --role="roles/storage.objectAdmin" + +echo +echo "bucket: $BUCKET" +echo "Bind on the deployment overlay (non-secret):" +echo " ECOMM_OBJECTSTORE_KIND=gcs" +echo " ECOMM_OBJECTSTORE_BUCKET=$BUCKET" +``` + +- [ ] **Step 3: Write `SKILL.md`** — frontmatter (name `provision-bucket`, description mirroring provision-datastore's: stand up the per-env private media bucket SD-0002 §6.3.1; idempotent; never public; grants the VM SA objectAdmin scoped to the bucket; bucket name is non-secret config) + the same Step-0 gate (scaffold-gcp-project + deployment.toml validate), Step-1 params reuse (`read-deployment-params.py` → NAME/PROJECT/GCLOUD_CONFIG/ZONE; derive REGION from ZONE), Step-2 run the script, Step-3 bind the overlay (`ECOMM_OBJECTSTORE_KIND` + `ECOMM_OBJECTSTORE_BUCKET` in deployment.toml `[overlay]`, re-import), Step-4 teaching contract (what exists: a private regional bucket, the VM SA can read/write it, no public endpoint). + +- [ ] **Step 4: SPEC §6.6b note** — add a short section in `launch-app/SPEC.md` after §6.6a documenting `provision-bucket` as the media-bucket sibling of `provision-datastore` (trigger: ecomm SD-0002 SLICE-7), same three invariants (engine needs no awareness — the bucket name is plain overlay config, not even a secret; idempotent; private by default). + +- [ ] **Step 5: Commit + PR + merge (engineering repo)** + +```bash +git add launch-app/skills/provision-bucket launch-app/SPEC.md +git commit -m "feat(launch-app): provision-bucket gesture — per-env private media bucket (§6.6b; ecomm SD-0002 SLICE-7)" +git push -u origin provision-bucket-gesture +# open + merge the PR via the Gitea API (issues/PR token), then return to the ecomm worktree +``` + +--- + +## Task 13: telemetry/docs/config glue + version bump + +**Files:** +- Modify: `deployment.toml` (`[overlay]`) +- Modify: `docs/products-domain.md`, `docs/OPERATIONS.md` +- Modify: `VERSION`, `frontend/package.json` (+ lockfile) + +- [ ] **Step 1: deployment.toml overlay** — add under `[overlay]`: + +```toml +ECOMM_OBJECTSTORE_KIND = "gcs" +ECOMM_OBJECTSTORE_BUCKET = "wiggleverse-ecomm-ppe-media" +``` + +(No secret — GCS auth is the VM service-account ADC. `public_base_url()` reuses the existing `APP_URL`.) + +- [ ] **Step 2: products-domain.md** — document the image pipeline (objectstore port, platform/images bar = 500px, the fetch phase + SSRF guard + bounds + thread model + resume), hosted-URL recognition (INV-12 over images), and **re-target the draft-blob seam** note: "BYTEA draft blob → objectstore" is *deferred past SLICE-7* (drafts remain BYTEA; the media bucket holds only product-images this slice). Update the "complete shortcut" seam: confirm now inserts `applying` → sets `fetching_images`/`complete`; the fetch task fills `image_progress`/`image_counts`/`image_outcomes`. + +- [ ] **Step 3: OPERATIONS.md** — add: the **`provision-bucket` operator gesture** (run launch-app `provision-bucket` before the first SLICE-7 PPE deploy; bind the overlay), the SLICE-7 telemetry events (TEL-4 `image_phase_completed`, TEL-5 `image_phase_recovered`), and **RB-3** (image phase stuck): how to read it — runs stuck in `fetching_images` are visible in the import history; a restart triggers the recovery scan (TEL-5); ALR-3 fires on >2 h stuck or ≥3 recoveries. Note ALR-3 is provisioned as a gcloud log-based alert at deploy (Task 14), mirroring ALR-2. + +- [ ] **Step 4: Version bump** — `VERSION` → `0.7.0`; `frontend/package.json` version → `0.7.0`; refresh lockfile (`cd frontend && npm install`). + +- [ ] **Step 5: Commit** + +```bash +git add deployment.toml docs/products-domain.md docs/OPERATIONS.md VERSION frontend/package.json frontend/package-lock.json +git commit -m "docs(products): image pipeline ops + seams (DOC-1/DOC-4); overlay bucket config; v0.7.0" +``` + +--- + +## Task 14: integration gate, ship through the pipeline + +**Files:** none new — verification, merge, deploy. + +- [ ] **Step 1: Extend the bootstrap test (INV-1 over images)** — in `backend/tests/test_bootstrap.py` (or the existing bootstrap test), add: fresh DB → sign-up → storefront → import `sample.csv` → export → re-import = all-unchanged still holds with the image columns present. (No image host needed; sample.csv's images stay `pending`/never-fetched → export keeps source_url → round-trips.) Run that test green. + +- [ ] **Step 2: Full local gate** + +Run: `bash scripts/check.sh` +Expected: lint-imports (layering kept), ~180+ backend pytest green, frontend typecheck+build, vitest green. + +Run: `bash scripts/e2e.sh` +Expected: 7/7 E2E green. + +- [ ] **Step 3: Open the PR (ecomm), self-review, merge** + +```bash +git push -u origin worktree-slice-7-images +# open PR via Gitea API; body cites SD-0002 §7.2 SLICE-7, PUC-7, INV-12/16/18, closes nothing +# (issue #13 stays open until SLICE-8). Merge after green local gates. +``` + +Tag the merge commit: `v0.7.0` and `release/` (PST) on the deployed ref (per `ppe-before-prod` §9.1). + +- [ ] **Step 4: Provision the bucket (operator gesture) + ALR-3** + +Hand the operator the `provision-bucket` command to run (it creates `wiggleverse-ecomm-ppe-media`, grants the VM SA objectAdmin, sets the lifecycle rule). Then provision **ALR-3** as a gcloud log-based alert in project `wiggleverse-ecomm`, mirroring SLICE-5's ALR-2: a log metric over `image_phase_recovered`/stuck-run signal + the existing email channel + an alert policy (condition: run in `fetching_images` > 2 h, or `image_phase_recovered` ≥ 3× for one run). These are operator-run/infra gestures (flotilla-only-provisioning + §8.4 dedicated config). + +- [ ] **Step 5: Deploy to PPE + verify (§9)** + +```bash +CLOUDSDK_ACTIVE_CONFIG_NAME=wiggleverse-ecomm flotilla-core deploy ecomm +``` + +Expected: 9/9 green. Then verify on `https://ecomm-ppe.wiggleverse.org`: `/healthz` reports `0.7.0`; sign in; import a small CSV with a public image URL; watch the run detail progress to a terminal status; confirm the served rendition loads (`/api/products/images/{id}/detail`) and the round-trip (export → re-import) is all-unchanged. Re-run `bash scripts/e2e.sh` is localhost-only; PPE verification is the manual walk (E2E-against-PPE is the §10.6 gap). + +- [ ] **Step 6: Finalize** — invoke `wgl-session-finalize` (archives this plan to the content repo `plans/`, marks SD-0002 §13 Q-3 resolved, updates memory with the SLICE-7 ship + the SLICE-8 next-goal, publishes the transcript, tears down the worktree). + +--- + +## Self-review notes (coverage vs §7.2 SLICE-7 DoD) + +- **`platform/objectstore` (local + GCS)** → Task 2. **`platform/images`** → Task 1. +- **Per-env media bucket via `provision-bucket`** → Task 12 (built) + Task 14 (operator runs). +- **Fetch task: SSRF guard, bounds, restart recovery** → Task 7 (+ wiring Task 8). +- **Image-serving endpoint** → Task 9. **INV-16 in force** → Task 9 (render paths take keys; route authorizes). +- **Hosted-URL resolution in diff + export (INV-12 over images)** → Tasks 3/4/5 + extended property test (Task 5). +- **Run-detail images section; Products-page notice band** → Task 10. +- **PUC-7 acceptance; E2E `e2e_image_outcomes`** → Task 11. **fetch-resume + SSRF integration** → Task 7. +- **TEL-4/5** → Task 7 (`image_phase_completed` / `image_phase_recovered`). **ALR-3** → Task 14. +- **DOC-1 updated (+ bucket gesture in operator guide)** → Task 13. **Q-3 pinned** → decisions block + Task 13/14. +- **Version 0.7.0 shipped through merge + PPE** → Tasks 13/14.