From 5d5341b29be7e5c68e07c898c4e75ebdaf3a1173 Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Fri, 12 Jun 2026 00:30:08 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20confirm=20=E2=86=92=20fetchin?= =?UTF-8?q?g=5Fimages=20+=20post-commit=20fetch=20scheduling=20+=20startup?= =?UTF-8?q?=20recovery=20(SD-0002=20=C2=A76.5.3/=C2=A76.9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- backend/app/domains/products/service.py | 4 +++- backend/app/main.py | 17 ++++++++++++++++- backend/tests/test_products_service.py | 20 ++++++++++++++++++++ 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/backend/app/domains/products/service.py b/backend/app/domains/products/service.py index fe377e4..b2dfa6c 100644 --- a/backend/app/domains/products/service.py +++ b/backend/app/domains/products/service.py @@ -132,11 +132,13 @@ def confirm_draft(conn: psycopg.Connection, storefront_id: int, account_id: int, 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="complete", + 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: diff --git a/backend/app/main.py b/backend/app/main.py index 71ddd98..20c47be 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -11,6 +11,7 @@ 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 @@ -24,6 +25,7 @@ from pydantic import BaseModel from app.domains import accounts, products, storefronts 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 @@ -117,13 +119,22 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None = @asynccontextmanager async def lifespan(app: FastAPI): - app.state.pool = db.open_pool(dsn) + app.state.pool = db.open_pool(dsn, max_size=10) 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) @@ -316,6 +327,10 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None = 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}") diff --git a/backend/tests/test_products_service.py b/backend/tests/test_products_service.py index 542b6ad..335d60e 100644 --- a/backend/tests/test_products_service.py +++ b/backend/tests/test_products_service.py @@ -220,3 +220,23 @@ def test_tel2_emitted_on_confirm(migrated_conn, merchant, caplog, telemetry_prop products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"]) events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"] assert any(e["event"] == "import_run_completed" and e["added"] == 2 for e in events) + + +def test_confirm_marks_run_fetching_images_when_images_present(migrated_conn, merchant): + csv = b"Handle,Title,Image Src\nlamp,Lamp,https://m.example/a.png\n" + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "c.csv", csv) + run_id = products.confirm_draft( + migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"]) + run = products.get_run(migrated_conn, merchant["storefront_id"], run_id) + assert run["status"] == "fetching_images" + assert run["image_progress"]["total"] >= 1 + + +def test_confirm_marks_run_complete_when_no_images(migrated_conn, merchant): + csv = b"Handle,Title,Vendor\nlamp,Lamp,Acme\n" + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "c.csv", csv) + run_id = products.confirm_draft( + migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"]) + assert products.get_run(migrated_conn, merchant["storefront_id"], run_id)["status"] == "complete"