# 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.