feat(products): import_validate → draft, preview records, discard + TEL-1

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app/domains/products/__init__.py

@@ -16,9 +16,11 @@ from .errors import (
     RunNotFound,
 )
 from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
+from .service import discard_draft, get_draft, get_draft_records, import_validate
 
 __all__ = [
     "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
     "PreviewStale", "NothingToApply", "RunNotFound",
     "MAX_DATA_ROWS", "MAX_FILE_BYTES",
+    "import_validate", "get_draft", "get_draft_records", "discard_draft",
 ]

backend/app/domains/products/service.py

@@ -0,0 +1,88 @@
+"""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 datetime import datetime, timezone
+
+import psycopg
+
+from app.platform import telemetry
+
+from . import codec, diff, repo, validate
+from .errors import DraftExpired, DraftNotFound
+
+
+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()

backend/app/platform/telemetry.py

@@ -0,0 +1,13 @@
+"""Structured log-event telemetry (SD-0002 §9.1). One JSON object per event on the
+`ecomm.telemetry` logger — counts and durations only; never file names, URLs,
+catalog content, or secret bytes (§6.3-handbook)."""
+from __future__ import annotations
+
+import json
+import logging
+
+_logger = logging.getLogger("ecomm.telemetry")
+
+
+def emit(event: str, **fields: object) -> None:
+    _logger.info(json.dumps({"event": event, **fields}, sort_keys=True, default=str))

backend/tests/test_products_service.py

@@ -0,0 +1,117 @@
+"""products service — drafts: validate/preview/discard (PUC-2/3/3a/5a; INV-11)."""
+import json
+import logging
+
+import psycopg
+import pytest
+
+from app.domains import products
+from app.platform import db
+
+GOOD_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,18.00\nstar-tee,Star Tee,Acme,24.00\n"
+
+
+@pytest.fixture()
+def migrated_conn(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn)
+        yield conn
+
+
+@pytest.fixture()
+def merchant(migrated_conn):
+    acct = migrated_conn.execute(
+        "INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0]
+    sf = migrated_conn.execute(
+        "INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0]
+    migrated_conn.execute(
+        "INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
+    migrated_conn.commit()
+    return {"account_id": acct, "storefront_id": sf}
+
+
+def test_import_validate_creates_draft_with_summary(migrated_conn, merchant):
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    assert draft["dialect"] == "canonical"
+    assert draft["summary"] == {"adds": 2, "updates": 0, "unchanged": 0, "errors": 0}
+    assert draft["expires_at"]
+
+
+def test_validate_writes_nothing_to_catalog_inv11(migrated_conn, merchant):
+    products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 0
+    assert migrated_conn.execute("SELECT count(*) FROM variant").fetchone()[0] == 0
+
+
+def test_file_rejection_leaves_no_draft(migrated_conn, merchant):
+    with pytest.raises(products.FileRejected):
+        products.import_validate(
+            migrated_conn, merchant["storefront_id"], merchant["account_id"], "bad.csv",
+            b"Vendor,Price\nAcme,1\n")
+    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0
+
+
+def test_records_paging_and_kind_filter(migrated_conn, merchant):
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    recs = products.get_draft_records(migrated_conn, merchant["storefront_id"], draft["id"])
+    assert [r["handle"] for r in recs] == ["moon-mug", "star-tee"]
+    adds = products.get_draft_records(
+        migrated_conn, merchant["storefront_id"], draft["id"], kind="add", limit=1)
+    assert len(adds) == 1 and adds[0]["kind"] == "add"
+
+
+def test_discard_deletes_no_trace_puc3a(migrated_conn, merchant):
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"])
+    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0
+    products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"])  # idempotent
+
+
+def test_draft_scoped_to_storefront_inv14(migrated_conn, merchant):
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    other_sf = migrated_conn.execute(
+        "INSERT INTO storefront (name) VALUES ('Other') RETURNING id").fetchone()[0]
+    migrated_conn.commit()
+    with pytest.raises(products.DraftNotFound):
+        products.get_draft(migrated_conn, other_sf, draft["id"])
+
+
+def test_expired_draft_raises_and_lazily_deletes(migrated_conn, merchant):
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    migrated_conn.execute(
+        "UPDATE import_draft SET expires_at = now() - interval '1 minute' WHERE id = %s",
+        (draft["id"],))
+    migrated_conn.commit()
+    with pytest.raises(products.DraftExpired):
+        products.get_draft(migrated_conn, merchant["storefront_id"], draft["id"])
+    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0
+
+
+@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_tel1_emitted(migrated_conn, merchant, caplog, telemetry_propagation):
+    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
+        products.import_validate(
+            migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
+    events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
+    assert any(
+        e["event"] == "import_draft_created" and e["adds"] == 2 and e["row_count"] == 2
+        and "duration_ms" in e and e["unknown_columns_count"] == 0
+        for e in events
+    )