feat(products): streamed export service + TEL-3 + EmptyCatalog (PUC-9, §9.1)

backend/app/domains/products/__init__.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from .errors import (
     DraftExpired,
     DraftNotFound,
+    EmptyCatalog,
     FileRejected,
     NothingToApply,
     PreviewStale,
@@ -21,6 +22,7 @@ from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
 from .service import (
     confirm_draft,
     discard_draft,
+    export_catalog,
     get_draft,
     get_draft_records,
     get_run,
@@ -34,8 +36,8 @@ SAMPLE_CSV_PATH = Path(__file__).parent / "sample.csv"
 
 __all__ = [
     "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
-    "PreviewStale", "NothingToApply", "RunNotFound",
+    "PreviewStale", "NothingToApply", "RunNotFound", "EmptyCatalog",
     "MAX_DATA_ROWS", "MAX_FILE_BYTES", "SAMPLE_CSV_PATH",
     "import_validate", "get_draft", "get_draft_records", "discard_draft",
-    "confirm_draft", "list_runs", "get_run", "summary",
+    "confirm_draft", "list_runs", "get_run", "summary", "export_catalog",
 ]

backend/app/domains/products/errors.py

@@ -35,3 +35,7 @@ class NothingToApply(ProductsError):
 
 class RunNotFound(ProductsError):
     """No such import run for this storefront."""
+
+
+class EmptyCatalog(ProductsError):
+    """PUC-9: nothing to export (no products, or none matching the status filter)."""

backend/app/domains/products/service.py

@@ -7,14 +7,22 @@ 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 telemetry
 
-from . import codec, diff, repo, validate
+from . import codec, diff, repo, serialize, validate
-from .errors import DraftExpired, DraftNotFound, NothingToApply, PreviewStale, RunNotFound
+from .errors import (
+    DraftExpired,
+    DraftNotFound,
+    EmptyCatalog,
+    NothingToApply,
+    PreviewStale,
+    RunNotFound,
+)
 
 
 def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int,
@@ -218,6 +226,34 @@ def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict:
     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)
+        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 {

backend/tests/test_products_export.py

@@ -8,7 +8,7 @@ import psycopg
 import pytest
 
 from app.domains import products
-from app.domains.products import repo, serialize
+from app.domains.products import repo
 from app.platform import db
 
 CSV = (
@@ -54,3 +54,47 @@ def test_export_status_filter(migrated_conn, merchant):
     assert [p.handle for p in active] == ["active-mug"]
     draft = repo.export_catalog(migrated_conn, merchant["storefront_id"], "draft")
     assert [p.handle for p in draft] == ["draft-tee"]
+
+
+@pytest.fixture()
+def telemetry_propagation():
+    """create_app() sets propagate=False on the parent "ecomm" logger
+    (main._ensure_app_logging), which hides ecomm.telemetry records from caplog's
+    root-logger handler whenever an API test ran first. Restore propagation here."""
+    lg = logging.getLogger("ecomm")
+    prior = lg.propagate
+    lg.propagate = True
+    yield
+    lg.propagate = prior
+
+
+def test_export_streams_canonical_csv(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)
+    text = "".join(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+    rows = list(csv.DictReader(io.StringIO(text)))
+    assert {r["Handle"] for r in rows} == {"active-mug", "draft-tee"}
+    assert rows[0]["Handle"] == "active-mug"  # sorted by handle
+
+
+def test_export_empty_raises(migrated_conn, merchant):
+    # No catalog at all → empty.
+    with pytest.raises(products.EmptyCatalog):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+
+
+def test_export_empty_after_filter_raises(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)  # only active + draft exist
+    with pytest.raises(products.EmptyCatalog):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "archived"))
+
+
+def test_tel3_emitted(migrated_conn, merchant, caplog, telemetry_propagation):
+    _seed(migrated_conn, merchant)
+    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+    events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
+    exported = [e for e in events if e["event"] == "catalog_exported"]
+    assert len(exported) == 1
+    assert exported[0]["product_count"] == 2
+    assert exported[0]["status_filter"] == "all"
+    assert "duration_ms" in exported[0]