From 87a3d4435941ec7c6dc7584e9daa83f80600be36 Mon Sep 17 00:00:00 2001 From: obroccolio Date: Sat, 12 Sep 2026 13:28:11 +0800 Subject: [PATCH] fix: persist product archive cover and provenance --- backend/service/app.py | 11 +-- backend/service/product_archive_service.py | 52 ++++++----- backend/service/product_archive_store.py | 78 ++++++++++++++++- backend/service/schemas.py | 6 +- backend/tests/test_product_archive.py | 95 +++++++++++++++++++++ backend/tests/test_product_archive_store.py | 56 ++++++++++++ 6 files changed, 262 insertions(+), 36 deletions(-) diff --git a/backend/service/app.py b/backend/service/app.py index cc0f3d8..e6e7176 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -1110,11 +1110,6 @@ async def create_product_version( except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - source_job_ids = [ - str(source.get("source_job_id") or "") - for source in version.metadata.get("wordcloud_sources", []) - if isinstance(source, dict) - ] rows = product_archive_store._fetchall( "SELECT * FROM product_wordcloud_archives WHERE version_id = ? ORDER BY created_at, archive_id", (version.version_id,), @@ -1126,10 +1121,12 @@ async def create_product_version( "version_id": row["version_id"], "archive_path": row["archive_path"], "created_at": datetime.fromisoformat(row["created_at"]), - "source_job_id": source_job_id, + "source_job_id": row["source_job_id"], + "source_asset_id": row["source_asset_id"], + "db_checksum": row["db_checksum"], "db_path": row["archive_path"], } - for source_job_id, row in zip(source_job_ids, rows, strict=True) + for row in rows ] return ProductVersionArchiveResponse( **version.model_dump(), diff --git a/backend/service/product_archive_service.py b/backend/service/product_archive_service.py index 3187ff7..5a0a7ed 100644 --- a/backend/service/product_archive_service.py +++ b/backend/service/product_archive_service.py @@ -68,12 +68,9 @@ class ProductArchiveService: resolved.append((source, db_path)) return resolved - def _rollback_metadata(self, version_id: str) -> None: - # ProductArchiveStore deliberately keeps its public API small. These rows - # all belong to the just-created version and can be removed safely here. - self.store._execute("DELETE FROM product_wordcloud_archives WHERE version_id = ?", (version_id,)) - self.store._execute("DELETE FROM product_images WHERE version_id = ?", (version_id,)) - self.store._execute("DELETE FROM product_versions WHERE version_id = ?", (version_id,)) + @staticmethod + def _move_staging(staging_dir: Path, final_dir: Path) -> None: + staging_dir.replace(final_dir) def archive_version( self, @@ -97,14 +94,13 @@ class ProductArchiveService: product_dir = self.archive_root / product_id staging_dir = product_dir / f".{uuid.uuid4().hex}.staging" - version: ProductVersionRecord | None = None final_dir: Path | None = None try: staging_dir.mkdir(parents=True, exist_ok=False) preview_path = staging_dir / "design-preview.png" preview_path.write_bytes(preview_bytes) - snapshots: list[tuple[WordcloudSource, Path]] = [] + snapshots: list[tuple[WordcloudSource, Path, str]] = [] for index, (source, source_db) in enumerate(source_dbs, start=1): snapshot_path = ( staging_dir @@ -112,37 +108,39 @@ class ProductArchiveService: / f"{index:02d}-{source.source_job_id}" / "word_locations.sqlite" ) - copy_word_locations_snapshot(source_db, snapshot_path) - snapshots.append((source, snapshot_path)) + checksum = copy_word_locations_snapshot(source_db, snapshot_path) + snapshots.append((source, snapshot_path, checksum)) - version = self.store.create_version( + version_id = f"ver_{uuid.uuid4().hex}" + target_dir = product_dir / version_id + self._move_staging(staging_dir, target_dir) + final_dir = target_dir + preview = final_dir / "design-preview.png" + archive_rows = [ + { + "archive_path": str(final_dir / snapshot_path.relative_to(staging_dir)), + "source_job_id": source.source_job_id, + "source_asset_id": source.asset_id, + "db_checksum": checksum, + } + for source, snapshot_path, checksum in snapshots + ] + version, _, _ = self.store.write_archive_version( product_id, + version_id=version_id, metadata={ "document": document, "wordcloud_count": len(snapshots), "wordcloud_sources": [ {"asset_id": source.asset_id, "source_job_id": source.source_job_id} - for source, _ in snapshots + for source, _, _ in snapshots ], }, + preview_path=str(preview), + archives=archive_rows, now=now, ) - target_dir = product_dir / version.version_id - staging_dir.replace(target_dir) - final_dir = target_dir - - final_preview = final_dir / "design-preview.png" - self.store.add_image( - product_id, version.version_id, image_path=str(final_preview), image_type="design_preview", now=now - ) - for _, snapshot_path in snapshots: - final_snapshot = final_dir / snapshot_path.relative_to(staging_dir) - self.store.add_wordcloud_archive( - product_id, version.version_id, archive_path=str(final_snapshot), now=now - ) return version except Exception: - if version is not None: - self._rollback_metadata(version.version_id) shutil.rmtree(final_dir or staging_dir, ignore_errors=True) raise diff --git a/backend/service/product_archive_store.py b/backend/service/product_archive_store.py index 108fab3..d828f75 100644 --- a/backend/service/product_archive_store.py +++ b/backend/service/product_archive_store.py @@ -54,7 +54,8 @@ class ProductArchiveStore: status TEXT NOT NULL DEFAULT 'active', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, - purge_after TEXT + purge_after TEXT, + cover_image_id TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_products_source_external_id ON products(source, external_product_id); @@ -77,6 +78,7 @@ class ProductArchiveStore: version_id TEXT NOT NULL, image_path TEXT NOT NULL DEFAULT '', image_type TEXT NOT NULL DEFAULT '', + is_cover INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_product_images_product_id @@ -89,6 +91,9 @@ class ProductArchiveStore: product_id TEXT NOT NULL, version_id TEXT NOT NULL, archive_path TEXT NOT NULL DEFAULT '', + source_job_id TEXT NOT NULL DEFAULT '', + source_asset_id TEXT NOT NULL DEFAULT '', + db_checksum TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_product_wordcloud_archives_product_id @@ -110,6 +115,17 @@ class ProductArchiveStore: ON cleanup_records(purge_after); """ ) + self._ensure_column(conn, "products", "cover_image_id", "TEXT") + self._ensure_column(conn, "product_images", "is_cover", "INTEGER NOT NULL DEFAULT 0") + self._ensure_column(conn, "product_wordcloud_archives", "source_job_id", "TEXT NOT NULL DEFAULT ''") + self._ensure_column(conn, "product_wordcloud_archives", "source_asset_id", "TEXT NOT NULL DEFAULT ''") + self._ensure_column(conn, "product_wordcloud_archives", "db_checksum", "TEXT NOT NULL DEFAULT ''") + + @staticmethod + def _ensure_column(conn: sqlite3.Connection, table: str, column: str, declaration: str) -> None: + columns = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")} + if column not in columns: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {declaration}") def _execute(self, sql: str, parameters: tuple[Any, ...] = ()) -> None: with self._lock, self._connect() as conn: @@ -132,6 +148,7 @@ class ProductArchiveStore: created_at=datetime.fromisoformat(row["created_at"]), updated_at=datetime.fromisoformat(row["updated_at"]), purge_after=datetime.fromisoformat(row["purge_after"]) if row["purge_after"] else None, + cover_image_id=row["cover_image_id"], ) def upsert_product(self, product: ProductInput) -> ProductRecord: @@ -198,6 +215,65 @@ class ProductArchiveStore: ) return record + def _before_archive_commit(self) -> None: + """Test seam: runs inside the archive metadata transaction before commit.""" + + def write_archive_version( + self, + product_id: str, + version_id: str, + metadata: dict[str, Any], + preview_path: str, + archives: list[dict[str, str]], + now: datetime | None = None, + ) -> tuple[ProductVersionRecord, ProductImageRecord, list[ProductWordcloudArchiveRecord]]: + """Persist one complete archive version and make its preview the sole cover.""" + created_at = now or _now() + version = ProductVersionRecord( + version_id=version_id, product_id=product_id, metadata=metadata, created_at=created_at + ) + image = ProductImageRecord( + image_id=f"img_{uuid.uuid4().hex}", product_id=product_id, version_id=version_id, + image_path=preview_path, image_type="design_preview", is_cover=True, created_at=created_at, + ) + records = [ + ProductWordcloudArchiveRecord( + archive_id=f"wca_{uuid.uuid4().hex}", product_id=product_id, version_id=version_id, + archive_path=item["archive_path"], source_job_id=item["source_job_id"], + source_asset_id=item["source_asset_id"], db_checksum=item["db_checksum"], created_at=created_at, + ) + for item in archives + ] + with self._lock, self._connect() as conn: + product = conn.execute("SELECT product_id FROM products WHERE product_id = ?", (product_id,)).fetchone() + if product is None: + raise ValueError(f"unknown product_id: {product_id}") + conn.execute("BEGIN") + conn.execute( + "INSERT INTO product_versions (version_id, product_id, version, metadata, created_at) VALUES (?, ?, ?, ?, ?)", + (version.version_id, version.product_id, version.version, json.dumps(version.metadata, ensure_ascii=False), version.created_at.isoformat()), + ) + conn.execute( + "UPDATE product_images SET is_cover = 0 WHERE product_id = ? AND is_cover = 1", (product_id,) + ) + conn.execute( + """INSERT INTO product_images (image_id, product_id, version_id, image_path, image_type, is_cover, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (image.image_id, image.product_id, image.version_id, image.image_path, image.image_type, 1, image.created_at.isoformat()), + ) + for record in records: + conn.execute( + """INSERT INTO product_wordcloud_archives + (archive_id, product_id, version_id, archive_path, source_job_id, source_asset_id, db_checksum, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + (record.archive_id, record.product_id, record.version_id, record.archive_path, + record.source_job_id, record.source_asset_id, record.db_checksum, record.created_at.isoformat()), + ) + conn.execute("UPDATE products SET cover_image_id = ?, updated_at = ? WHERE product_id = ?", (image.image_id, created_at.isoformat(), product_id)) + self._before_archive_commit() + conn.commit() + return version, image, records + def add_image( self, product_id: str, version_id: str, image_path: str = "", image_type: str = "", now: datetime | None = None, diff --git a/backend/service/schemas.py b/backend/service/schemas.py index 977fd9b..91e5ee8 100644 --- a/backend/service/schemas.py +++ b/backend/service/schemas.py @@ -47,6 +47,7 @@ class ProductRecord(BaseModel): created_at: datetime updated_at: datetime purge_after: datetime | None = None + cover_image_id: str | None = None class ProductVersionRecord(BaseModel): @@ -63,6 +64,7 @@ class ProductImageRecord(BaseModel): version_id: str image_path: str = "" image_type: str = "" + is_cover: bool = False created_at: datetime @@ -71,11 +73,13 @@ class ProductWordcloudArchiveRecord(BaseModel): product_id: str version_id: str archive_path: str = "" + source_job_id: str = "" + source_asset_id: str = "" + db_checksum: str = "" created_at: datetime class ProductWordcloudArchiveResponse(ProductWordcloudArchiveRecord): - source_job_id: str db_path: str diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index aed1b07..edb11ba 100644 --- a/backend/tests/test_product_archive.py +++ b/backend/tests/test_product_archive.py @@ -55,6 +55,7 @@ def product_archive_client(tmp_path, monkeypatch): monkeypatch.setattr(service_app, "product_archive_store", store, raising=False) client = TestClient(service_app.app) client.archive_root = archive_root + client.product_store = store return client @@ -89,6 +90,15 @@ def prepared_wordcloud_job(tmp_path, monkeypatch): ) +def archive_product_version(client, product_id, document, preview=PNG_BYTES): + return client.post( + f"/api/products/{product_id}/versions", + data={"document_json": json.dumps(document)}, + files={"preview": ("design-preview.png", preview, "image/png")}, + headers=orders_auth_header(), + ) + + def test_scanner_keeps_only_visible_wordcloud_assets_and_deduplicates(): document = { "layers": [{"id": "shown", "visible": True}, {"id": "hidden", "visible": False}], @@ -252,3 +262,88 @@ def test_archive_version_handles_document_without_wordcloud_sources(product_arch assert response.status_code == 201 assert response.json()["wordcloud_count"] == 0 + + +def test_archive_version_rejects_invalid_png_bytes_with_png_mime(product_archive_client): + product = create_product(product_archive_client) + + response = archive_product_version(product_archive_client, product["product_id"], {"elements": []}, b"\x89PNG\r\n\x1a\nnot-an-image") + + assert response.status_code == 400 + + +def test_archive_version_persists_each_multi_source_provenance(product_archive_client, prepared_wordcloud_job, monkeypatch): + second_db = prepared_wordcloud_job.workspace / "output" / "second.sqlite" + with sqlite3.connect(second_db) as connection: + connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)") + second_asset = "asset_second" + second_dir = service_app._asset_dir(second_asset) + second_dir.mkdir(parents=True) + (second_dir / "meta.json").write_text(json.dumps({"type": "wordcloud", "job_id": "job-second"}), encoding="utf-8") + monkeypatch.setattr( + service_app, "_resolve_job_status", + lambda job_id: SimpleNamespace(status="success", artifacts={"db": str(prepared_wordcloud_job.db_path if job_id == "job-success" else second_db)}), + ) + product = create_product(product_archive_client) + response = archive_product_version(product_archive_client, product["product_id"], { + "elements": [ + {"type": "sticker", "assetId": prepared_wordcloud_job.asset_id}, + {"type": "sticker", "assetId": second_asset}, + ] + }) + + assert response.status_code == 201 + archives = {item["source_job_id"]: item for item in response.json()["wordcloud_archives"]} + assert archives["job-success"]["source_asset_id"] == prepared_wordcloud_job.asset_id + assert archives["job-second"]["source_asset_id"] == second_asset + assert archives["job-success"]["db_checksum"] == hashlib.sha256(Path(archives["job-success"]["db_path"]).read_bytes()).hexdigest() + assert archives["job-second"]["db_checksum"] == hashlib.sha256(Path(archives["job-second"]["db_path"]).read_bytes()).hexdigest() + + +def test_later_archive_version_becomes_current_cover_in_product_contract(product_archive_client): + product = create_product(product_archive_client) + first = archive_product_version(product_archive_client, product["product_id"], {"elements": []}) + first_cover = product_archive_client.get(f"/api/products/{product['product_id']}", headers=orders_auth_header()).json()["cover_image_id"] + second = archive_product_version(product_archive_client, product["product_id"], {"elements": []}) + detail = product_archive_client.get(f"/api/products/{product['product_id']}", headers=orders_auth_header()).json() + listed = product_archive_client.get("/api/products", headers=orders_auth_header()).json()[0] + + assert first.status_code == second.status_code == 201 + assert detail["cover_image_id"] == listed["cover_image_id"] + assert detail["cover_image_id"] != first_cover + + +def test_product_routes_require_orders_auth(product_archive_client): + product = create_product(product_archive_client) + requests = [ + product_archive_client.post("/api/products", json={"name": "未授权", "source": "manual"}), + product_archive_client.get("/api/products"), + product_archive_client.get(f"/api/products/{product['product_id']}"), + product_archive_client.post(f"/api/products/{product['product_id']}/versions"), + product_archive_client.delete(f"/api/products/{product['product_id']}"), + product_archive_client.post(f"/api/products/{product['product_id']}/restore"), + ] + + assert [response.status_code for response in requests] == [403] * 6 + + +def test_archive_failure_during_final_move_removes_staging_files(product_archive_client, prepared_wordcloud_job, monkeypatch): + product = create_product(product_archive_client) + service = service_app._product_archive_service() + monkeypatch.setattr(service, "_move_staging", lambda *_: (_ for _ in ()).throw(RuntimeError("move failed"))) + + with pytest.raises(RuntimeError, match="move failed"): + service.archive_version(product["product_id"], prepared_wordcloud_job.document, PNG_BYTES) + + assert not list((product_archive_client.archive_root / product["product_id"]).iterdir()) + + +def test_archive_failure_during_metadata_transaction_removes_final_files(product_archive_client, prepared_wordcloud_job, monkeypatch): + product = create_product(product_archive_client) + monkeypatch.setattr(product_archive_client.product_store, "_before_archive_commit", lambda: (_ for _ in ()).throw(RuntimeError("metadata failed"))) + + with pytest.raises(RuntimeError, match="metadata failed"): + service_app._product_archive_service().archive_version(product["product_id"], prepared_wordcloud_job.document, PNG_BYTES) + + assert not list((product_archive_client.archive_root / product["product_id"]).iterdir()) + assert product_archive_client.product_store._fetchall("SELECT * FROM product_versions") == [] diff --git a/backend/tests/test_product_archive_store.py b/backend/tests/test_product_archive_store.py index 7494d5d..e28ea26 100644 --- a/backend/tests/test_product_archive_store.py +++ b/backend/tests/test_product_archive_store.py @@ -113,3 +113,59 @@ def test_version_bound_operations_reject_unknown_or_mismatched_versions(tmp_path store.add_image(second.product_id, version.version_id) with pytest.raises(ValueError): store.add_wordcloud_archive(second.product_id, version.version_id) + + +def test_archive_write_is_atomic_and_replaces_the_current_cover(tmp_path): + store = ProductArchiveStore(tmp_path / "service_products") + product = store.upsert_product(ProductInput("manual", None, "笔盒")) + + first, first_cover, first_archives = store.write_archive_version( + product_id=product.product_id, + version_id="ver_first", + metadata={}, + preview_path="/archive/first/design-preview.png", + archives=[{ + "archive_path": "/archive/first/word_locations.sqlite", + "source_job_id": "job-first", + "source_asset_id": "asset-first", + "db_checksum": "first-checksum", + }], + now=NOW, + ) + second, second_cover, second_archives = store.write_archive_version( + product_id=product.product_id, + version_id="ver_second", + metadata={}, + preview_path="/archive/second/design-preview.png", + archives=[], + now=NOW + timedelta(seconds=1), + ) + + assert first_cover.is_cover is True + assert first_archives[0].source_job_id == "job-first" + assert first_archives[0].source_asset_id == "asset-first" + assert first_archives[0].db_checksum == "first-checksum" + assert store.get_product(product.product_id).cover_image_id == second_cover.image_id + assert store._fetchone("SELECT is_cover FROM product_images WHERE image_id = ?", (first_cover.image_id,))["is_cover"] == 0 + assert store._fetchone("SELECT is_cover FROM product_images WHERE image_id = ?", (second_cover.image_id,))["is_cover"] == 1 + assert first.version_id != second.version_id + + +def test_archive_write_rolls_back_all_rows_when_the_transaction_fails(tmp_path, monkeypatch): + store = ProductArchiveStore(tmp_path / "service_products") + product = store.upsert_product(ProductInput("manual", None, "笔盒")) + monkeypatch.setattr(store, "_before_archive_commit", lambda: (_ for _ in ()).throw(RuntimeError("injected"))) + + with pytest.raises(RuntimeError, match="injected"): + store.write_archive_version( + product_id=product.product_id, + version_id="ver_failure", + metadata={}, + preview_path="/archive/failure/design-preview.png", + archives=[], + now=NOW, + ) + + assert store._fetchall("SELECT * FROM product_versions") == [] + assert store._fetchall("SELECT * FROM product_images") == [] + assert store.get_product(product.product_id).cover_image_id is None