chore(reset)!: strip storefront → network-service seed (v0.9.0)

Pivot ecomm from a generic multi-tenant Shopify-alternative storefront to the
commitment-commerce + cross-maker verified referral NETWORK direction
(ecomm-content/rfcs/network_strategy.md; OHM identity/relationality super-drafts).

Phase A of the reset — clear the decks:
- Strip the storefront surfaces with no carry-forward: the storefronts domain,
  the products import/export/image HTTP spine (service/imagefetch/repo + endpoints),
  the SPA frontend, and the Playwright e2e suite.
- Keep the reusable core: /healthz + /api/auth/* (accounts); the
  catalog-normalization library (codec/dialect/models/serialize/validate/diff/hosted —
  importable, no HTTP yet); platform/* infra. Migrations untouched (append-only).
- Reduce check.sh to backend-only (import-linter + pytest); trim dev.sh and the unused
  GCS overlay env. Repoint app.json/README/CLAUDE.md; bump VERSION 0.8.0 -> 0.9.0.

check.sh green: import-linter (main > domains > platform) KEPT, pytest passing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/domains/__init__.py

@@ -1,6 +1,6 @@
-"""domains layer — bounded contexts (accounts, storefronts).
+"""domains layer — bounded contexts (accounts, products).
 
-Empty in SLICE-1; the accounts domain lands in SLICE-2 and storefronts in SLICE-3.
-The package exists now so the layer contract (.importlinter) has a target and later
-slices add to a stable seam. Domains import only from app.platform, never upward.
+The accounts domain is the network-service seed's identity context. The products
+package is the catalog-normalization library (no HTTP surface yet). Domains import
+only from app.platform, never upward.
 """

backend/app/domains/products/__init__.py

@@ -1,14 +1,27 @@
-"""products domain — catalog + bulk CSV import/export (SD-0002 §6.2).
+"""products domain — catalog-normalization library (no HTTP surface yet).
 
-Owns the canonical row model, codec, validation, diff engine, and import
-drafts/runs. Storefront-scoped throughout (INV-14); upsert is the only mutation
-(INV-10). Imported via this package surface only.
+The network-seed reset stripped the storefront-scoped import/export/image spine
+(service/repo/imagefetch and their endpoints). What remains is the pure, DB-free
+catalog-normalization library: the canonical row model, the CSV codec, the Shopify
+dialect adapter, validation, the diff engine, and serialization. These submodules are
+importable as a library; nothing here owns persistence or an HTTP route.
 """
 from __future__ import annotations
 
 from pathlib import Path
 
-from .imagefetch import recover_incomplete_runs, run_image_phase
+from .codec import detect_dialect, parse_csv
+from .diff import (
+    CatalogImage,
+    CatalogProduct,
+    CatalogVariant,
+    DiffResult,
+    ImagePlan,
+    ProductPlan,
+    VariantPlan,
+    compute_diff,
+)
+from .dialect_shopify import is_shopify_header, map_shopify_header
 from .errors import (
     DraftExpired,
     DraftNotFound,
@@ -19,30 +32,37 @@ from .errors import (
     ProductsError,
     RunNotFound,
 )
-from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
-from .repo import image_for_serving
-from .service import (
-    confirm_draft,
-    discard_draft,
-    export_catalog,
-    get_draft,
-    get_draft_records,
-    get_run,
-    import_validate,
-    list_runs,
-    summary,
+from .models import (
+    MAX_DATA_ROWS,
+    MAX_FILE_BYTES,
+    CanonicalImage,
+    CanonicalProduct,
+    CanonicalVariant,
+    ParsedFile,
+    Row,
+    RowError,
 )
+from .serialize import catalog_to_csv
+from .validate import all_errors, build_products
 
-# DOC-3: the downloadable worked-example CSV the BFF serves at /api/products/sample.csv.
+# DOC-3: the downloadable worked-example CSV.
 SAMPLE_CSV_PATH = Path(__file__).parent / "sample.csv"
-# DOC-2: the column reference the BFF serves at /api/products/columns.md.
+# DOC-2: the column reference.
 COLUMNS_MD_PATH = Path(__file__).parent / "columns.md"
 
 __all__ = [
+    # errors
     "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
     "PreviewStale", "NothingToApply", "RunNotFound", "EmptyCatalog",
-    "MAX_DATA_ROWS", "MAX_FILE_BYTES", "SAMPLE_CSV_PATH", "COLUMNS_MD_PATH",
-    "import_validate", "get_draft", "get_draft_records", "discard_draft",
-    "confirm_draft", "list_runs", "get_run", "summary", "export_catalog",
-    "run_image_phase", "recover_incomplete_runs", "image_for_serving",
+    # models + limits
+    "MAX_DATA_ROWS", "MAX_FILE_BYTES", "Row", "RowError", "ParsedFile",
+    "CanonicalProduct", "CanonicalVariant", "CanonicalImage",
+    # codec + dialect
+    "parse_csv", "detect_dialect", "is_shopify_header", "map_shopify_header",
+    # validate + diff + serialize
+    "build_products", "all_errors", "compute_diff", "catalog_to_csv",
+    "CatalogProduct", "CatalogVariant", "CatalogImage",
+    "ProductPlan", "VariantPlan", "ImagePlan", "DiffResult",
+    # docs
+    "SAMPLE_CSV_PATH", "COLUMNS_MD_PATH",
 ]

backend/app/domains/products/imagefetch.py

@@ -1,146 +0,0 @@
-"""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 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
-    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)
-    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:
-    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
-    try:
-        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()
-    except Exception as exc:  # noqa: BLE001 — one bad image must never wedge the run (review fix)
-        with pool.connection() as conn:
-            repo.mark_image_failed(conn, image_id, f"processing error: {type(exc).__name__}")
-            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

backend/app/domains/products/repo.py

@@ -1,616 +0,0 @@
-"""products repo — the SQL layer for the import spine (SD-0002 §6.3 data model).
-
-Owns SQL only: the catalog snapshot the diff engine reads, import draft/run
-CRUD, and the apply primitives the confirm transaction calls. Business rules
-live in service.py and diff.py — nothing here validates, diffs, commits, or
-rolls back (the confirm flow runs the apply primitives inside its own
-transaction). Every catalog/draft/run query is storefront-scoped (INV-14).
-
-Dict payload conventions: functions feeding §6.4 API payloads (insert_draft,
-list_runs, get_run) return datetimes as `.isoformat()` strings; get_draft_row
-returns raw datetimes for the service's expiry check. TEXT[] columns bind/load
-as Python lists and NUMERIC loads as Decimal natively under psycopg 3.
-"""
-from __future__ import annotations
-
-import psycopg
-from psycopg import sql
-from psycopg.types.json import Jsonb
-
-from .diff import CatalogImage, CatalogProduct, CatalogVariant
-
-# ---------------------------------------------------------------------------
-# Catalog snapshot (diff input) + dashboard counts
-# ---------------------------------------------------------------------------
-
-
-def load_catalog(conn: psycopg.Connection, storefront_id: int) -> dict[str, CatalogProduct]:
-    """The storefront's full catalog, keyed by handle, in diff.py's snapshot shape."""
-    catalog: dict[str, CatalogProduct] = {}
-    by_id: dict[int, CatalogProduct] = {}
-    for row in conn.execute(
-        "SELECT id, handle, title, description_html, vendor, product_type,"
-        " google_product_category, tags, status, published,"
-        " option1_name, option2_name, option3_name"
-        " FROM product WHERE storefront_id = %s",
-        (storefront_id,),
-    ):
-        product = CatalogProduct(
-            id=row[0],
-            handle=row[1],
-            title=row[2],
-            option_names=(row[10], row[11], row[12]),
-            fields={
-                "title": row[2],
-                "description_html": row[3],
-                "vendor": row[4],
-                "product_type": row[5],
-                "google_product_category": row[6],
-                "tags": row[7],
-                "status": row[8],
-                "published": row[9],
-            },
-            variants=[],
-            images=[],
-        )
-        catalog[product.handle] = product
-        by_id[product.id] = product
-    for row in conn.execute(
-        "SELECT v.product_id, v.id, v.position,"
-        " v.option1_value, v.option2_value, v.option3_value,"
-        " v.sku, v.barcode, v.price, v.cost, v.weight, v.weight_unit,"
-        " v.volume, v.volume_unit, v.tax_id_1, v.tax_id_2,"
-        " v.inventory_tracker, v.inventory_qty, i.source_url"
-        " FROM variant v"
-        " JOIN product p ON p.id = v.product_id"
-        " LEFT JOIN product_image i ON i.id = v.image_id"
-        " WHERE p.storefront_id = %s"
-        " ORDER BY v.product_id, v.position, v.id",
-        (storefront_id,),
-    ):
-        by_id[row[0]].variants.append(
-            CatalogVariant(
-                id=row[1],
-                options=(row[3], row[4], row[5]),
-                position=row[2],
-                fields={
-                    "sku": row[6],
-                    "barcode": row[7],
-                    "price": row[8],
-                    "cost": row[9],
-                    "weight": row[10],
-                    "weight_unit": row[11],
-                    "volume": row[12],
-                    "volume_unit": row[13],
-                    "tax_id_1": row[14],
-                    "tax_id_2": row[15],
-                    "inventory_tracker": row[16],
-                    "inventory_qty": row[17],
-                    "variant_image": row[18],
-                },
-            )
-        )
-    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])
-        )
-    return catalog
-
-
-_EXPORT_STATUSES = ("all", "active", "draft", "archived")
-
-
-def export_catalog(
-    conn: psycopg.Connection, storefront_id: int, status_filter: str
-) -> list[CatalogProduct]:
-    """The storefront's catalog as an ordered snapshot list, optionally filtered
-    by product status (PUC-9). Reuses load_catalog's snapshot builder; the
-    catalog fits in memory (≤5k rows, INV-18). Ordered by handle for a stable,
-    deterministic export."""
-    catalog = load_catalog(conn, storefront_id)
-    products = sorted(catalog.values(), key=lambda p: p.handle)
-    if status_filter and status_filter != "all":
-        products = [p for p in products if p.fields.get("status") == status_filter]
-    return products
-
-
-def product_count(conn: psycopg.Connection, storefront_id: int) -> int:
-    return conn.execute(
-        "SELECT count(*) FROM product WHERE storefront_id = %s", (storefront_id,)
-    ).fetchone()[0]
-
-
-def image_problem_count(conn: psycopg.Connection, storefront_id: int) -> int:
-    return conn.execute(
-        "SELECT count(*) FROM product_image i"
-        " JOIN product p ON p.id = i.product_id"
-        " WHERE p.storefront_id = %s"
-        " AND i.status IN ('rejected_low_res', 'rejected_not_image', 'failed')",
-        (storefront_id,),
-    ).fetchone()[0]
-
-
-def latest_run_id(conn: psycopg.Connection, storefront_id: int) -> int | None:
-    row = conn.execute(
-        "SELECT id FROM import_run WHERE storefront_id = %s"
-        " ORDER BY created_at DESC, id DESC LIMIT 1",
-        (storefront_id,),
-    ).fetchone()
-    return row[0] if row else None
-
-
-# ---------------------------------------------------------------------------
-# Import drafts (preview server side, INV-11)
-# ---------------------------------------------------------------------------
-
-
-def insert_draft(
-    conn: psycopg.Connection,
-    storefront_id: int,
-    account_id: int,
-    file_name: str,
-    dialect: str,
-    file_bytes: bytes,
-    summary: dict,
-    records: list,
-    fingerprint: str,
-    unknown_columns: list[str],
-) -> dict:
-    """Create a draft (expires in 1 hour); returns the §6.4 draft payload."""
-    row = conn.execute(
-        "INSERT INTO import_draft"
-        " (storefront_id, account_id, file_name, dialect, file_bytes,"
-        "  summary, records, fingerprint, unknown_columns, expires_at)"
-        " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, now() + interval '1 hour')"
-        " RETURNING id, expires_at",
-        (
-            storefront_id,
-            account_id,
-            file_name,
-            dialect,
-            file_bytes,
-            Jsonb(summary),
-            Jsonb(records),
-            fingerprint,
-            unknown_columns,
-        ),
-    ).fetchone()
-    return {
-        "id": row[0],
-        "file_name": file_name,
-        "dialect": dialect,
-        "summary": summary,
-        "unknown_columns": unknown_columns,
-        "expires_at": row[1].isoformat(),
-    }
-
-
-def get_draft_row(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> dict | None:
-    row = conn.execute(
-        "SELECT id, storefront_id, account_id, file_name, dialect, file_bytes,"
-        " summary, records, fingerprint, unknown_columns, expires_at, created_at"
-        " FROM import_draft WHERE id = %s AND storefront_id = %s",
-        (draft_id, storefront_id),
-    ).fetchone()
-    if row is None:
-        return None
-    columns = (
-        "id",
-        "storefront_id",
-        "account_id",
-        "file_name",
-        "dialect",
-        "file_bytes",
-        "summary",
-        "records",
-        "fingerprint",
-        "unknown_columns",
-        "expires_at",
-        "created_at",
-    )
-    record = dict(zip(columns, row))
-    # BYTEA loads as memoryview; the service expects bytes.
-    record["file_bytes"] = bytes(record["file_bytes"])
-    return record
-
-
-def draft_records(
-    conn: psycopg.Connection,
-    storefront_id: int,
-    draft_id: int,
-    kind: str | None,
-    limit: int,
-    offset: int,
-) -> list[dict]:
-    """The draft's preview records, order-preserving, optionally filtered by kind."""
-    rows = conn.execute(
-        "SELECT rec FROM import_draft d,"
-        " jsonb_array_elements(d.records) WITH ORDINALITY AS r(rec, ord)"
-        " WHERE d.id = %(draft_id)s AND d.storefront_id = %(storefront_id)s"
-        " AND (%(kind)s::text IS NULL OR rec->>'kind' = %(kind)s)"
-        " ORDER BY ord LIMIT %(limit)s OFFSET %(offset)s",
-        {
-            "draft_id": draft_id,
-            "storefront_id": storefront_id,
-            "kind": kind,
-            "limit": limit,
-            "offset": offset,
-        },
-    ).fetchall()
-    return [row[0] for row in rows]
-
-
-def delete_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> None:
-    conn.execute(
-        "DELETE FROM import_draft WHERE id = %s AND storefront_id = %s",
-        (draft_id, storefront_id),
-    )
-
-
-def sweep_expired_drafts(conn: psycopg.Connection) -> None:
-    conn.execute("DELETE FROM import_draft WHERE expires_at < now()")
-
-
-# ---------------------------------------------------------------------------
-# Import runs (history, PUC-8)
-# ---------------------------------------------------------------------------
-
-_TERMINAL_RUN_STATUSES = ("complete", "complete_with_problems")
-
-
-def insert_run(
-    conn: psycopg.Connection,
-    storefront_id: int,
-    account_id: int,
-    file_name: str,
-    dialect: str,
-    added: int,
-    updated: int,
-    errored: int,
-    status: str,
-) -> int:
-    return conn.execute(
-        "INSERT INTO import_run"
-        " (storefront_id, account_id, file_name, dialect,"
-        "  products_added, products_updated, rows_errored, status, completed_at)"
-        " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, CASE WHEN %s THEN now() END)"
-        " RETURNING id",
-        (
-            storefront_id,
-            account_id,
-            file_name,
-            dialect,
-            added,
-            updated,
-            errored,
-            status,
-            status in _TERMINAL_RUN_STATUSES,
-        ),
-    ).fetchone()[0]
-
-
-def insert_run_errors(conn: psycopg.Connection, run_id: int, errors: list[dict]) -> None:
-    """Record per-row errors (RowError.as_json shape: line/column/message)."""
-    if not errors:
-        return
-    with conn.cursor() as cur:
-        cur.executemany(
-            "INSERT INTO import_run_error (run_id, line_number, column_name, message)"
-            " VALUES (%s, %s, %s, %s)",
-            [(run_id, e["line"], e["column"], e["message"]) for e in errors],
-        )
-
-
-_RUN_SELECT = (
-    "SELECT r.id, r.file_name, r.dialect, r.created_at, r.completed_at, r.status,"
-    " a.email, r.products_added, r.products_updated, r.rows_errored"
-    " FROM import_run r JOIN account a ON a.id = r.account_id"
-)
-
-
-def _run_dict(row: tuple) -> dict:
-    return {
-        "id": row[0],
-        "file_name": row[1],
-        "dialect": row[2],
-        "created_at": row[3].isoformat(),
-        "completed_at": row[4].isoformat() if row[4] is not None else None,
-        "status": row[5],
-        "by": row[6],
-        "products_added": row[7],
-        "products_updated": row[8],
-        "rows_errored": row[9],
-    }
-
-
-def list_runs(conn: psycopg.Connection, storefront_id: int, limit: int, offset: int) -> list[dict]:
-    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
-
-
-def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict | None:
-    row = conn.execute(
-        _RUN_SELECT + " WHERE r.storefront_id = %s AND r.id = %s",
-        (storefront_id, run_id),
-    ).fetchone()
-    if row is None:
-        return None
-    run = _run_dict(row)
-    run["errors"] = [
-        {"line": line, "column": column, "message": message}
-        for line, column, message in conn.execute(
-            "SELECT line_number, column_name, message FROM import_run_error"
-            " WHERE run_id = %s ORDER BY line_number, id",
-            (run_id,),
-        )
-    ]
-    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)
-    return run
-
-
-# ---------------------------------------------------------------------------
-# Image fetch phase (SLICE-7, SD-0002 §6.5.4) — claim/mark/count/outcomes.
-# ---------------------------------------------------------------------------
-
-
-def pending_images_for_run(conn: psycopg.Connection, run_id: int) -> list[dict]:
-    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:
-    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]:
-    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 image_for_serving(conn: psycopg.Connection, storefront_id: int, image_id: int) -> dict | None:
-    """Storefront-scoped lookup for the serving route: status + rendition keys (INV-14)."""
-    row = conn.execute(
-        "SELECT i.status, i.key_original, i.key_thumb, i.key_card, i.key_detail"
-        " FROM product_image i JOIN product p ON p.id = i.product_id"
-        " WHERE i.id = %s AND p.storefront_id = %s",
-        (image_id, storefront_id),
-    ).fetchone()
-    if row is None:
-        return None
-    return {"status": row[0], "original": row[1], "thumb": row[2], "card": row[3], "detail": row[4]}
-
-
-def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]:
-    rows = conn.execute(
-        "SELECT p.handle, i.source_url, i.status, i.failure_reason,"
-        " 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
-    ]
-
-
-# ---------------------------------------------------------------------------
-# Apply primitives — called inside the confirm transaction (Task 8); no commits.
-# ---------------------------------------------------------------------------
-
-
-def insert_product(
-    conn: psycopg.Connection,
-    storefront_id: int,
-    handle: str,
-    resolved_fields: dict,
-    option_names: tuple[str | None, str | None, str | None],
-) -> int:
-    """INSERT with only the file-present fields; absent ones take column defaults."""
-    columns = ["storefront_id", "handle", "option1_name", "option2_name", "option3_name"]
-    values: list[object] = [storefront_id, handle, *option_names]
-    for field_name, value in resolved_fields.items():
-        columns.append(field_name)
-        values.append(value)
-    query = sql.SQL("INSERT INTO product ({}) VALUES ({}) RETURNING id").format(
-        sql.SQL(", ").join(sql.Identifier(c) for c in columns),
-        sql.SQL(", ").join(sql.Placeholder() for _ in columns),
-    )
-    return conn.execute(query, values).fetchone()[0]
-
-
-def update_product(conn: psycopg.Connection, product_id: int, changed_fields: dict) -> None:
-    if not changed_fields:
-        return
-    assignments = [
-        sql.SQL("{} = {}").format(sql.Identifier(f), sql.Placeholder())
-        for f in changed_fields
-    ]
-    query = sql.SQL("UPDATE product SET {}, updated_at = now() WHERE id = {}").format(
-        sql.SQL(", ").join(assignments), sql.Placeholder()
-    )
-    conn.execute(query, [*changed_fields.values(), product_id])
-
-
-def insert_variant(
-    conn: psycopg.Connection,
-    product_id: int,
-    position: int,
-    options: tuple[str | None, str | None, str | None],
-    resolved_fields: dict,
-    image_id: int | None,
-) -> int:
-    # variant_image is not a column — the caller translates it to image_id; position
-    # is the explicit param. Filter both defensively.
-    fields = {
-        k: v for k, v in resolved_fields.items() if k != "variant_image" and k != "position"
-    }
-    columns = [
-        "product_id",
-        "position",
-        "option1_value",
-        "option2_value",
-        "option3_value",
-        "image_id",
-    ]
-    values: list[object] = [product_id, position, *options, image_id]
-    for field_name, value in fields.items():
-        columns.append(field_name)
-        values.append(value)
-    query = sql.SQL("INSERT INTO variant ({}) VALUES ({}) RETURNING id").format(
-        sql.SQL(", ").join(sql.Identifier(c) for c in columns),
-        sql.SQL(", ").join(sql.Placeholder() for _ in columns),
-    )
-    return conn.execute(query, values).fetchone()[0]
-
-
-def update_variant(
-    conn: psycopg.Connection,
-    variant_id: int,
-    changed_fields: dict,
-    image_id: int | None | type(...) = ...,
-) -> None:
-    """Dynamic UPDATE; image_id's Ellipsis default means "don't touch image_id"."""
-    fields = {
-        k: v for k, v in changed_fields.items() if k != "variant_image"
-    }
-    assignments = [
-        sql.SQL("{} = {}").format(sql.Identifier(f), sql.Placeholder()) for f in fields
-    ]
-    values: list[object] = list(fields.values())
-    if image_id is not ...:
-        assignments.append(sql.SQL("image_id = {}").format(sql.Placeholder()))
-        values.append(image_id)
-    if not assignments:
-        return
-    query = sql.SQL("UPDATE variant SET {}, updated_at = now() WHERE id = {}").format(
-        sql.SQL(", ").join(assignments), sql.Placeholder()
-    )
-    conn.execute(query, [*values, variant_id])
-
-
-def get_or_create_image(
-    conn: psycopg.Connection,
-    product_id: int,
-    source_url: str,
-    position: int,
-    alt_text: str | None,
-    run_id: int,
-) -> int:
-    """Image identity within a product is source_url (§6.3); existing rows are
-    returned untouched — diff emits explicit image update entries for position/alt."""
-    row = conn.execute(
-        "SELECT id FROM product_image WHERE product_id = %s AND source_url = %s",
-        (product_id, source_url),
-    ).fetchone()
-    if row is not None:
-        return row[0]
-    return conn.execute(
-        "INSERT INTO product_image (product_id, source_url, position, alt_text, import_run_id)"
-        " VALUES (%s, %s, %s, %s, %s) RETURNING id",
-        (product_id, source_url, position, alt_text, run_id),
-    ).fetchone()[0]
-
-
-def update_image(conn: psycopg.Connection, image_id: int, changed_fields: dict) -> None:
-    """Subset of {position, alt_text}."""
-    if not changed_fields:
-        return
-    assignments = [
-        sql.SQL("{} = {}").format(sql.Identifier(f), sql.Placeholder())
-        for f in changed_fields
-    ]
-    query = sql.SQL("UPDATE product_image SET {} WHERE id = {}").format(
-        sql.SQL(", ").join(assignments), sql.Placeholder()
-    )
-    conn.execute(query, [*changed_fields.values(), image_id])

backend/app/domains/products/service.py

@@ -1,265 +0,0 @@
-"""products service — the import/export use-case orchestration (SD-0002 §6.5).
-
-Coordinates codec → validate → diff → repo; owns transaction boundaries (repo
-never commits). Preview is read-only against catalog tables (INV-11): validation
-writes exactly one row — the import_draft. TEL events per §9.1.
-"""
-from __future__ import annotations
-
-import time
-from collections.abc import Iterator
-from datetime import datetime, timezone
-
-import psycopg
-
-from app.platform import config, telemetry
-
-from . import codec, diff, repo, serialize, validate
-from .errors import (
-    DraftExpired,
-    DraftNotFound,
-    EmptyCatalog,
-    NothingToApply,
-    PreviewStale,
-    RunNotFound,
-)
-
-
-def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int,
-                    file_name: str, data: bytes) -> dict:
-    """Upload → validate → diff → persist draft (PUC-2/3; INV-11). Raises FileRejected."""
-    started = time.monotonic()
-    # Commit the sweep before parsing: a FileRejected mid-parse must not roll
-    # back expired-draft cleanup along with it.
-    repo.sweep_expired_drafts(conn)
-    conn.commit()
-    parsed = codec.parse_csv(data)
-    products = validate.build_products(parsed)
-    catalog = repo.load_catalog(conn, storefront_id)
-    diff_result = diff.compute_diff(catalog, products)
-    draft = repo.insert_draft(
-        conn, storefront_id, account_id, file_name, parsed.dialect, data,
-        diff_result.summary, diff_result.records, diff_result.fingerprint,
-        parsed.unknown_columns,
-    )
-    conn.commit()
-    telemetry.emit(
-        "import_draft_created",
-        storefront_id=storefront_id,
-        dialect=parsed.dialect,
-        row_count=len(parsed.rows),
-        adds=diff_result.summary["adds"],
-        updates=diff_result.summary["updates"],
-        unchanged=diff_result.summary["unchanged"],
-        errors=diff_result.summary["errors"],
-        unknown_columns_count=len(parsed.unknown_columns),
-        duration_ms=int((time.monotonic() - started) * 1000),
-    )
-    return draft
-
-
-def _live_draft_row(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> dict:
-    """The draft row if it exists and hasn't expired; expiry deletes lazily (§6.3)."""
-    row = repo.get_draft_row(conn, storefront_id, draft_id)
-    if row is None:
-        raise DraftNotFound()
-    if row["expires_at"] < datetime.now(timezone.utc):
-        repo.delete_draft(conn, storefront_id, draft_id)
-        conn.commit()
-        raise DraftExpired()
-    return row
-
-
-def get_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> dict:
-    """The §6.4 draft payload — never file_bytes or the full records list."""
-    row = _live_draft_row(conn, storefront_id, draft_id)
-    return {
-        "id": row["id"],
-        "file_name": row["file_name"],
-        "dialect": row["dialect"],
-        "summary": row["summary"],
-        "unknown_columns": row["unknown_columns"],
-        "expires_at": row["expires_at"].isoformat(),
-    }
-
-
-def get_draft_records(conn: psycopg.Connection, storefront_id: int, draft_id: int,
-                      kind: str | None = None, limit: int = 100, offset: int = 0) -> list[dict]:
-    """The draft's preview records, paged, optionally filtered by kind (PUC-3)."""
-    _live_draft_row(conn, storefront_id, draft_id)
-    return repo.draft_records(conn, storefront_id, draft_id, kind, limit, offset)
-
-
-def discard_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> None:
-    """Delete the draft, no trace kept; idempotent — an absent draft is fine (PUC-3a)."""
-    repo.delete_draft(conn, storefront_id, draft_id)
-    conn.commit()
-
-
-def confirm_draft(conn: psycopg.Connection, storefront_id: int, account_id: int,
-                  draft_id: int) -> int:
-    """Apply the previewed diff in one transaction (PUC-4; INV-10/11).
-
-    Everything is re-derived from the draft's stored file bytes against the live
-    catalog; a fingerprint mismatch means the catalog drifted since preview
-    (PreviewStale — the draft is kept so the merchant can re-validate). The apply
-    executes the typed plan compute_diff built alongside the preview records, so
-    what lands is exactly what the preview showed. rows_errored counts the
-    import_run_error rows recorded (one per RowError), which is what the run
-    detail's error table shows; the preview's errors tile counts error *products*.
-    """
-    started = time.monotonic()
-    row = _live_draft_row(conn, storefront_id, draft_id)
-    parsed = codec.parse_csv(row["file_bytes"])
-    products = validate.build_products(parsed)
-    catalog = repo.load_catalog(conn, storefront_id)
-    diff_result = diff.compute_diff(catalog, products)
-    if diff_result.fingerprint != row["fingerprint"]:
-        # Release the read snapshot; nothing written.
-        conn.rollback()
-        raise PreviewStale()
-    summary_counts = diff_result.summary
-    if summary_counts["adds"] + summary_counts["updates"] == 0:
-        # Release the read snapshot; nothing written.
-        conn.rollback()
-        raise NothingToApply()
-    error_rows = [
-        error.as_json()
-        for plan in diff_result.plan if plan.kind == "error"
-        for error in plan.canonical.errors
-    ]
-    try:
-        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()
-    except Exception as exc:
-        conn.rollback()
-        telemetry.emit(
-            "import_apply_failed",
-            draft_id=draft_id,
-            storefront_id=storefront_id,
-            error_class=type(exc).__name__,
-        )
-        raise
-    telemetry.emit(
-        "import_run_completed",
-        run_id=run_id,
-        storefront_id=storefront_id,
-        added=summary_counts["adds"],
-        updated=summary_counts["updates"],
-        errored=len(error_rows),
-        duration_ms=int((time.monotonic() - started) * 1000),
-    )
-    return run_id
-
-
-def _apply_product_plan(conn: psycopg.Connection, storefront_id: int,
-                        plan: diff.ProductPlan, run_id: int) -> None:
-    """Execute one product's plan inside the confirm transaction (no commits here)."""
-    if plan.kind == "add":
-        # Title is a canonical attribute, not a fields{} entry — non-error
-        # products always carry one (validate guarantees it).
-        product_fields = {"title": plan.canonical.title}
-        product_fields.update(diff.resolved_product_fields(plan.canonical))
-        product_id = repo.insert_product(
-            conn, storefront_id, plan.canonical.handle, product_fields, plan.canonical.option_names
-        )
-        image_ids: dict[str, int] = {}
-    elif plan.kind == "update":
-        product_id = plan.catalog.id
-        repo.update_product(conn, product_id, plan.product_changes)
-        image_ids = {image.source_url: image.id for image in plan.catalog.images}
-    else:
-        return
-
-    # Images first, so variants' variant_image URLs resolve to ids: validate puts
-    # every variant_image URL into canonical.images, so each URL is in either the
-    # catalog map (existing image) or the adds below.
-    for image_plan in plan.image_plans:
-        if image_plan.kind == "add":
-            image_ids[image_plan.source_url] = repo.get_or_create_image(
-                conn, product_id, image_plan.source_url, image_plan.position,
-                image_plan.alt_text, run_id,
-            )
-        else:
-            repo.update_image(conn, image_plan.image_id, image_plan.changes)
-
-    for variant_plan in plan.variant_plans:
-        if variant_plan.kind == "add":
-            fields = diff.resolved_variant_fields(variant_plan.canonical, variant_plan.file_order)
-            # diff time resolved any cleared position to file order; the
-            # file_order fallback covers an absent position column.
-            position = fields.get("position") or variant_plan.file_order
-            url = fields.get("variant_image")
-            image_id = image_ids[url] if url else None
-            repo.insert_variant(
-                conn, product_id, position, variant_plan.canonical.options, fields, image_id
-            )
-        elif "variant_image" in variant_plan.changes:
-            url = variant_plan.changes["variant_image"]
-            repo.update_variant(
-                conn, variant_plan.catalog_id, variant_plan.changes,
-                image_id=image_ids[url] if url else None,
-            )
-        else:
-            repo.update_variant(conn, variant_plan.catalog_id, variant_plan.changes)
-
-
-def list_runs(conn: psycopg.Connection, storefront_id: int,
-              limit: int = 50, offset: int = 0) -> list[dict]:
-    """The storefront's import history, newest first (PUC-8)."""
-    return repo.list_runs(conn, storefront_id, limit, offset)
-
-
-def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict:
-    """One run's §6.4 detail payload, errors included."""
-    run = repo.get_run(conn, storefront_id, run_id)
-    if run is None:
-        raise RunNotFound()
-    return run
-
-
-def export_catalog(
-    conn: psycopg.Connection, storefront_id: int, status_filter: str
-) -> Iterator[str]:
-    """Stream the storefront's catalog as canonical CSV (PUC-9; INV-12 codec).
-
-    Read-only: builds the snapshot, then streams the serializer over it. TEL-3
-    is emitted once the stream is exhausted, with the product count and elapsed
-    time. Raises EmptyCatalog before yielding anything if the (filtered) catalog
-    is empty, so the BFF can answer 409 cleanly with no partial body.
-    """
-    started = time.monotonic()
-    snapshot = repo.export_catalog(conn, storefront_id, status_filter)
-    if not snapshot:
-        raise EmptyCatalog()
-
-    def _stream() -> Iterator[str]:
-        yield from serialize.catalog_to_csv(snapshot, base_url=config.public_base_url())
-        telemetry.emit(
-            "catalog_exported",
-            storefront_id=storefront_id,
-            status_filter=status_filter,
-            product_count=len(snapshot),
-            duration_ms=int((time.monotonic() - started) * 1000),
-        )
-
-    return _stream()
-
-
-def summary(conn: psycopg.Connection, storefront_id: int) -> dict:
-    """The products dashboard counts (§6.4)."""
-    return {
-        "product_count": repo.product_count(conn, storefront_id),
-        "image_problem_count": repo.image_problem_count(conn, storefront_id),
-        "latest_run_id": repo.latest_run_id(conn, storefront_id),
-    }

backend/app/domains/storefronts/__init__.py

@@ -1,19 +0,0 @@
-"""storefronts domain — storefront entity, membership, the one-storefront guard.
-
-Owns INV-4 (one storefront per account is a service-layer rule) and INV-5 (tenant rows
-carry storefront_id). Knows nothing about identity (that is the accounts domain). Imported
-via this package surface only (§6.2).
-"""
-from __future__ import annotations
-
-from .errors import AlreadyOwnsStorefront, StorefrontsError
-from .models import Storefront
-from .service import create_storefront, storefront_for
-
-__all__ = [
-    "Storefront",
-    "StorefrontsError",
-    "AlreadyOwnsStorefront",
-    "create_storefront",
-    "storefront_for",
-]

backend/app/domains/storefronts/errors.py

@@ -1,10 +0,0 @@
-"""storefronts domain errors."""
-from __future__ import annotations
-
-
-class StorefrontsError(Exception):
-    """Base for storefronts-domain errors."""
-
-
-class AlreadyOwnsStorefront(StorefrontsError):
-    """INV-4: the account already has its one storefront (PUC-7)."""

backend/app/domains/storefronts/models.py

@@ -1,10 +0,0 @@
-"""Storefront record — the §6.3 entity as the domain returns it."""
-from __future__ import annotations
-
-from dataclasses import dataclass
-
-
-@dataclass(frozen=True)
-class Storefront:
-    id: int
-    name: str

backend/app/domains/storefronts/service.py

@@ -1,56 +0,0 @@
-"""storefronts — the storefront + membership service (SD-0001 §6.5).
-
-Owns the storefront entity, the account<->storefront membership (INV-5), the entry-routing
-answer (storefront_for), and INV-4's one-storefront guard — the single deletable check that
-makes one-per-account an MVP rule, not a schema law. Never mints identity: the BFF passes
-account_id and email down.
-"""
-from __future__ import annotations
-
-import psycopg
-
-from .errors import AlreadyOwnsStorefront
-from .models import Storefront
-
-
-def _default_name(email: str) -> str:
-    """§6.3: a blank name stores a generated default — never NULL (corpus 14.01.0026)."""
-    return f"{email.split('@', 1)[0]}'s storefront"
-
-
-def storefront_for(conn: psycopg.Connection, account_id: int) -> Storefront | None:
-    """The entry-routing answer (§6.5): which storefront, if any, this account has."""
-    row = conn.execute(
-        "SELECT s.id, s.name FROM storefront s"
-        " JOIN storefront_membership m ON m.storefront_id = s.id"
-        " WHERE m.account_id = %s ORDER BY m.created_at LIMIT 1",
-        (account_id,),
-    ).fetchone()
-    return Storefront(id=row[0], name=row[1]) if row else None
-
-
-def create_storefront(
-    conn: psycopg.Connection, account_id: int, email: str, name: str | None
-) -> Storefront:
-    """Create the account's one storefront + owner membership (PUC-4; INV-4, INV-5).
-
-    Guard + insert run as one atomic unit: a transaction-scoped advisory lock keyed by
-    account_id serializes concurrent creates for the same account, so the second of two
-    racing requests sees the first's committed membership and is refused (§6.5).
-    """
-    conn.execute("SELECT pg_advisory_xact_lock(%s)", (account_id,))
-    existing = storefront_for(conn, account_id)
-    if existing is not None:
-        conn.rollback()  # release the advisory lock; nothing was written
-        raise AlreadyOwnsStorefront()
-    final_name = (name or "").strip() or _default_name(email)
-    sf_id = conn.execute(
-        "INSERT INTO storefront (name) VALUES (%s) RETURNING id", (final_name,)
-    ).fetchone()[0]
-    conn.execute(
-        "INSERT INTO storefront_membership (account_id, storefront_id, role)"
-        " VALUES (%s, %s, 'owner')",
-        (account_id, sf_id),
-    )
-    conn.commit()
-    return Storefront(id=sf_id, name=final_name)

backend/app/main.py

@@ -1,32 +1,31 @@
-"""ecomm backend — FastAPI app factory + the REST BFF.
+"""ecomm backend — FastAPI app factory + the network-service seed BFF.
 
-SLICE-1 mounted /healthz; SLICE-2 adds the /api/auth/* identity endpoints (§6.4). The BFF
-translates HTTP <-> domain calls and owns no business logic (INV-6): every rule lives in
-the accounts domain. create_app() opens the pool, self-migrates (INV-1, INV-7), and builds
-the configured mailer (INV-8) at startup. SLICE-3 adds POST /api/storefronts and feeds the
-_storefront_for seam from the storefronts domain. SLICE-5 adds the /api/products/* import
-spine (SD-0002 §6.4): each endpoint is a gate + one products-domain call + error mapping.
+This is the pivot to a network-service seed: a minimal FastAPI surface carrying only
+/healthz and the /api/auth/* identity endpoints (§6.4), backed by the accounts domain.
+The BFF translates HTTP <-> domain calls and owns no business logic (INV-6): every rule
+lives in the accounts domain. create_app() opens the pool, self-migrates (INV-1, INV-7),
+and builds the configured mailer (INV-8 — auth needs the mailer) at startup.
+
+The catalog-normalization library (app.domains.products: codec/dialect/models/serialize/
+validate/diff) is kept as importable modules but has no HTTP surface yet — the storefront
+and products import/export/image spine was stripped in the network-seed reset.
 """
 from __future__ import annotations
 
 import logging
 import sys
-from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any
 
 import psycopg
-from fastapi import Depends, FastAPI, File, Query, Response, UploadFile
-from fastapi import Path as ApiPath
-from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
-from fastapi.staticfiles import StaticFiles
+from fastapi import Depends, FastAPI, Response
+from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
-from app.domains import accounts, products, storefronts
+from app.domains import accounts
 from app.platform import config, db
 from app.platform import mailer as mailer_mod
-from app.platform import objectstore as objectstore_mod
 from app.platform.deps import SESSION_COOKIE, get_conn, get_mailer, get_session
 from app.platform.mailer import Mailer
 from app.platform import session as session_mod
@@ -50,41 +49,11 @@ class VerifyBody(BaseModel):
     code: str
 
 
-class CreateStorefrontBody(BaseModel):
-    name: str | None = None
-
-
 def _error(status: int, code: str, message: str, **extra: Any) -> JSONResponse:
     """The shared §6.4 error envelope: {"error": {"code", "message", ...}}."""
     return JSONResponse(status_code=status, content={"error": {"code": code, "message": message, **extra}})
 
 
-def _storefront_for(conn: psycopg.Connection, account: accounts.Account) -> dict | None:
-    """The entry-routing answer: which storefront, if any, this account has (§6.5)."""
-    sf = storefronts.storefront_for(conn, account.id)
-    return {"id": sf.id, "name": sf.name} if sf else None
-
-
-def _merchant_gate(
-    conn: psycopg.Connection, sess: dict | None
-) -> JSONResponse | tuple[accounts.Account, storefronts.Storefront]:
-    """The shared /api/products/* gate: a signed-in account that has its storefront.
-
-    Returns the (account, storefront) pair, or the ready-to-return error response —
-    401 with no session, 404 before the storefront exists (INV-14: every products
-    call is storefront-scoped, so there is nothing to address yet).
-    """
-    if sess is None:
-        return _error(401, "unauthenticated", "You are not signed in.")
-    account = accounts.get_account(conn, sess["account_id"])
-    if account is None:
-        return _error(401, "unauthenticated", "You are not signed in.")
-    sf = storefronts.storefront_for(conn, account.id)
-    if sf is None:
-        return _error(404, "no_storefront", "Create your storefront first.")
-    return account, sf
-
-
 def _ensure_app_logging() -> None:
     """Surface the app's own `ecomm.*` INFO logs on stderr (idempotent).
 
@@ -114,7 +83,7 @@ def _set_session_cookie(response: Response, account: accounts.Account) -> None:
     )
 
 
-def create_app(database_url: str | None = None, static_dir: str | Path | None = None) -> FastAPI:
+def create_app(database_url: str | None = None) -> FastAPI:
     _ensure_app_logging()
     dsn = database_url or config.database_url()
 
@@ -124,18 +93,9 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
         with app.state.pool.connection() as conn:
             db.migrate(conn)  # self-migrate at startup (INV-1, INV-7)
         app.state.mailer = mailer_mod.build_mailer(config.mailer_kind())  # INV-8
-        app.state.objectstore = objectstore_mod.build_objectstore(config.objectstore_kind())
-        app.state.image_allow_private = config.image_fetch_allow_private()
-        app.state.image_runner = ThreadPoolExecutor(max_workers=2, thread_name_prefix="img-run")
-        # §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, app.state.image_allow_private,
-        )
         try:
             yield
         finally:
-            app.state.image_runner.shutdown(wait=False)
             app.state.pool.close()
 
     app = FastAPI(title="ecomm", version=_APP_VERSION, lifespan=lifespan)
@@ -180,7 +140,7 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
 
     @app.post("/api/auth/verify")
     def verify(body: VerifyBody, conn: psycopg.Connection = Depends(get_conn)):
-        """Verify a code, start a session, and answer entry routing (§6.4/§6.5)."""
+        """Verify a code and start a session (§6.4)."""
         try:
             account, created = accounts.verify(conn, body.email, body.code)
         except accounts.InvalidEmail:
@@ -196,7 +156,6 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
             return _error(400, "code_exhausted", "Too many attempts — request a fresh code.")
         payload = {
             "account": {"email": account.email},
-            "storefront": _storefront_for(conn, account),
             "created": created,
         }
         resp = JSONResponse(status_code=200, content=payload)
@@ -205,13 +164,13 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
 
     @app.get("/api/auth/me")
     def me(conn: psycopg.Connection = Depends(get_conn), sess: dict | None = Depends(get_session)):
-        """The signed-in account + entry-routing answer, or 401 (§6.4/§6.5)."""
+        """The signed-in account, or 401 (§6.4)."""
         if sess is None:
             return _error(401, "unauthenticated", "You are not signed in.")
         account = accounts.get_account(conn, sess["account_id"])
         if account is None:
             return _error(401, "unauthenticated", "You are not signed in.")
-        return {"account": {"email": account.email}, "storefront": _storefront_for(conn, account)}
+        return {"account": {"email": account.email}}
 
     @app.post("/api/auth/logout")
     def logout():
@@ -220,251 +179,6 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
         resp.delete_cookie(SESSION_COOKIE, path="/")
         return resp
 
-    @app.post("/api/storefronts")
-    def create_storefront(
-        body: CreateStorefrontBody,
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Create the account's one storefront (§6.4; PUC-4, INV-4/PUC-7 on refusal)."""
-        if sess is None:
-            return _error(401, "unauthenticated", "You are not signed in.")
-        account = accounts.get_account(conn, sess["account_id"])
-        if account is None:
-            return _error(401, "unauthenticated", "You are not signed in.")
-        try:
-            sf = storefronts.create_storefront(conn, account.id, account.email, body.name)
-        except storefronts.AlreadyOwnsStorefront:
-            return _error(
-                409, "already_owns_storefront",
-                "Your account already has its storefront — ecomm is one storefront per account today.",
-            )
-        return JSONResponse(status_code=201, content={"id": sf.id, "name": sf.name})
-
-    @app.post("/api/products/imports")
-    async def import_upload(
-        file: UploadFile = File(...),
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Upload a CSV → validated import draft (§6.4; PUC-2, PUC-5/5a on rejection)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        account, sf = gate
-        data = await file.read()
-        if len(data) > products.MAX_FILE_BYTES:
-            return _error(413, "file_too_large", "This file is larger than 10 MB.")
-        try:
-            draft = products.import_validate(conn, sf.id, account.id, file.filename or "upload.csv", data)
-        except products.FileRejected as exc:
-            return _error(400, exc.code, exc.message)
-        return JSONResponse(status_code=201, content=draft)
-
-    @app.get("/api/products/imports/drafts/{draft_id}")
-    def get_import_draft(
-        draft_id: int,
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """One draft's preview payload — summary, never the file bytes (§6.4; PUC-3)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        try:
-            return products.get_draft(conn, sf.id, draft_id)
-        except products.DraftNotFound:
-            return _error(404, "not_found", "No such import preview.")
-        except products.DraftExpired:
-            return _error(410, "draft_expired", "This preview expired — upload the file again.")
-
-    @app.get("/api/products/imports/drafts/{draft_id}/records")
-    def get_import_draft_records(
-        draft_id: int,
-        kind: str | None = Query(default=None, pattern="^(add|update|unchanged|error)$"),
-        limit: int = Query(default=100, ge=1, le=500),
-        offset: int = Query(default=0, ge=0),
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """The draft's per-product preview records, paged + kind-filtered (§6.4; PUC-3)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        try:
-            records = products.get_draft_records(conn, sf.id, draft_id, kind, limit, offset)
-        except products.DraftNotFound:
-            return _error(404, "not_found", "No such import preview.")
-        except products.DraftExpired:
-            return _error(410, "draft_expired", "This preview expired — upload the file again.")
-        return {"records": records}
-
-    @app.post("/api/products/imports/drafts/{draft_id}/confirm")
-    def confirm_import_draft(
-        draft_id: int,
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Apply the previewed diff as one import run (§6.4; PUC-4, INV-10/11)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        account, sf = gate
-        try:
-            run_id = products.confirm_draft(conn, sf.id, account.id, draft_id)
-        except products.DraftNotFound:
-            return _error(404, "not_found", "No such import preview.")
-        except products.DraftExpired:
-            return _error(410, "draft_expired", "This preview expired — upload the file again.")
-        except products.PreviewStale:
-            return _error(
-                409, "preview_stale",
-                "Your catalog changed since this preview — upload the file again.",
-            )
-        except products.NothingToApply:
-            return _error(
-                409, "nothing_to_apply",
-                "Nothing to change — your catalog already matches this file.",
-            )
-        app.state.image_runner.submit(
-            products.run_image_phase, app.state.pool, app.state.objectstore,
-            run_id, app.state.image_allow_private,
-        )
-        return JSONResponse(status_code=201, content={"run_id": run_id})
-
-    @app.delete("/api/products/imports/drafts/{draft_id}")
-    def discard_import_draft(
-        draft_id: int,
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Discard the draft, no trace kept; idempotent (§6.4; PUC-3a)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        products.discard_draft(conn, sf.id, draft_id)
-        return Response(status_code=204)
-
-    @app.get("/api/products/imports/runs")
-    def list_import_runs(
-        limit: int = Query(default=50, ge=1, le=200),
-        offset: int = Query(default=0, ge=0),
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """The storefront's import history, newest first (§6.4; PUC-8)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        return {"runs": products.list_runs(conn, sf.id, limit, offset)}
-
-    @app.get("/api/products/imports/runs/{run_id}")
-    def get_import_run(
-        run_id: int,
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """One run's detail payload, errors included (§6.4; PUC-8)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        try:
-            return products.get_run(conn, sf.id, run_id)
-        except products.RunNotFound:
-            return _error(404, "not_found", "No such import run.")
-
-    @app.get("/api/products/summary")
-    def products_summary(
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """The products dashboard counts (§6.4)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        return products.summary(conn, sf.id)
-
-    @app.get("/api/products/export")
-    def export_products(
-        status: str = Query(default="all", pattern="^(all|active|draft|archived)$"),
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Stream the catalog as canonical CSV, optionally status-filtered (§6.4; PUC-9)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        try:
-            stream = products.export_catalog(conn, sf.id, status)
-        except products.EmptyCatalog:
-            return _error(409, "empty_catalog", "There are no products to export.")
-        return StreamingResponse(
-            stream,
-            media_type="text/csv",
-            headers={"content-disposition": 'attachment; filename="ecomm-products-export.csv"'},
-        )
-
-    _RENDITION_CT = {"thumb": "image/webp", "card": "image/webp",
-                     "detail": "image/webp", "original": "application/octet-stream"}
-
-    @app.get("/api/products/images/{image_id}/{rendition}")
-    def serve_product_image(
-        image_id: int,
-        rendition: str = ApiPath(pattern="^(original|thumb|card|detail)$"),
-        conn: psycopg.Connection = Depends(get_conn),
-        sess: dict | None = Depends(get_session),
-    ):
-        """Serve a hosted image rendition, storefront-authorized + immutable cache (§6.4, INV-16)."""
-        gate = _merchant_gate(conn, sess)
-        if isinstance(gate, JSONResponse):
-            return gate
-        _account, sf = gate
-        rec = products.image_for_serving(conn, sf.id, image_id)
-        if rec is None:
-            return _error(404, "not_found", "No such image.")
-        if rec["status"] != "fetched":
-            return _error(409, "not_fetched", "This image has not been fetched yet.")
-        key = rec[rendition]
-        if not key:
-            return _error(404, "not_found", "No such rendition.")
-        try:
-            data = app.state.objectstore.get(key)
-        except objectstore_mod.ObjectNotFound:
-            return _error(404, "not_found", "No such image.")
-        return Response(content=data, media_type=_RENDITION_CT[rendition],
-                        headers={"Cache-Control": "public, max-age=31536000, immutable"})
-
-    @app.get("/api/products/sample.csv")
-    def products_sample_csv():
-        """The DOC-3 worked-example CSV. Documentation, so no auth gate (§6.4)."""
-        return PlainTextResponse(
-            products.SAMPLE_CSV_PATH.read_text(),
-            media_type="text/csv",
-            headers={"content-disposition": 'attachment; filename="ecomm-products-sample.csv"'},
-        )
-
-    @app.get("/api/products/columns.md")
-    def products_columns_md():
-        """The DOC-2 column reference. Documentation, so no auth gate (§6.4)."""
-        return PlainTextResponse(
-            products.COLUMNS_MD_PATH.read_text(),
-            media_type="text/markdown; charset=utf-8",
-        )
-
-    # Deployed topology (launch-app SPEC §2): nginx proxies everything here, so the
-    # backend serves the built SPA. Mounted LAST so /healthz and /api/* win. In dev the
-    # dist dir doesn't exist (Vite serves the frontend) and the mount is skipped.
-    spa_dir = Path(static_dir) if static_dir is not None else _REPO_ROOT / "frontend" / "dist"
-    if (spa_dir / "index.html").is_file():
-        app.mount("/", StaticFiles(directory=spa_dir, html=True), name="spa")
-
     return app