diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index d5c6b06..4efe87b 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -13,7 +13,7 @@ from functools import cache from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection from pymongo import UpdateOne -from pymongo.errors import PyMongoError +from pymongo.errors import BulkWriteError, PyMongoError from monitor_utils.constants import ( ALERTS_COLLECTION_NAME, @@ -78,10 +78,11 @@ async def add_products_to_database( Rewrites current fields on each upsert. Prices are recorded in a pastPrices array of price/date entries; a price already present in the array is never recorded again, so each distinct price keeps only its - first-seen timestamp. New products and price changes - insert an alert referencing the product; alerts insert only after the - product write succeeds so a failed upsert cannot produce an orphaned - alert. + first-seen timestamp. New products and price changes upsert an alert + referencing the product, deduplicated so a product holds at most one + unclaimed alert regardless of concurrent category or page tasks + detecting it; alerts write only after the product write succeeds so a + failed upsert cannot produce an orphaned alert. Args: products: Products to upsert. @@ -103,7 +104,7 @@ async def add_products_to_database( existing = {str(doc["_id"]): doc async for doc in cursor} ops = [] - alerts = [] + alert_ops = [] now = datetime.now(timezone.utc) for product, doc_id in zip(products, doc_ids): document = convert_product_to_mongo(product) @@ -114,7 +115,14 @@ async def add_products_to_database( price_changed = new_price is not None and new_price != old_price set_on_insert = {"discoveredDate": now} if existing_doc is None or price_changed: - alerts.append(build_alert(monitor_name, document, now)) + alert = build_alert(monitor_name, document, now) + alert_ops.append( + UpdateOne( + {"product_id": doc_id, "claimed": False}, + {"$setOnInsert": alert}, + upsert=True, + ) + ) update = {"$set": document, "$setOnInsert": set_on_insert} recorded_prices = extract_recorded_prices(existing_doc) @@ -125,10 +133,8 @@ async def add_products_to_database( try: await collection.bulk_write(ops, ordered=False) - if alerts: - await get_collection(ALERTS_COLLECTION_NAME).insert_many( - alerts, ordered=False - ) + if alert_ops: + await write_alert_ops(alert_ops) except PyMongoError: LOGGER.exception( "failed to add products to database", @@ -137,6 +143,33 @@ async def add_products_to_database( raise +async def write_alert_ops(alert_ops: list[UpdateOne]): + """Write alert upserts, swallowing duplicate-key races. + + Concurrent tasks upserting an alert for the same product can both pass + the upsert's match phase and race the insert; the partial unique index + on unclaimed product_id makes the loser fail with a duplicate-key + error, which is the dedupe working as intended and is ignored. Any + other write error re-raises. + + Args: + alert_ops: Alert upsert operations to write. + """ + + try: + await get_collection(ALERTS_COLLECTION_NAME).bulk_write( + alert_ops, ordered=False + ) + except BulkWriteError as error: + real_errors = [ + write_error + for write_error in error.details["writeErrors"] + if write_error.get("code") != 11000 + ] + if real_errors: + raise + + def extract_recorded_prices(existing_doc: dict | None) -> set[float]: """Return the set of prices already recorded in a document's pastPrices. @@ -159,32 +192,48 @@ def build_alert(monitor_name: str, document: dict, now: datetime) -> dict: Args: monitor_name: Monitor that produced the product, used by the consumer to resolve the owning collection and thresholds. - document: Product document supplying the id and priority percentage. + document: Product document supplying the id and pricing for the + priority calculation. now: Detection timestamp stored as createdAt. Returns: - The alert document for insertion. + The alert document for upsert via $setOnInsert. """ - percentage = document.get("percentage") or 0 - data = { - "monitor": monitor_name, - "product_id": document["_id"], - "priority": min(percentage // 10, 10), - "claimed": False, - "attempts": 0, - "createdAt": now, - } return { "monitor": monitor_name, "product_id": document["_id"], - "priority": min(percentage // 10, 10), + "priority": calculate_priority(document), "claimed": False, "attempts": 0, "createdAt": now, } +def calculate_priority(document: dict) -> int: + """Derive an alert's priority bucket from its discount percentage. + + Always recalculated from current and original price so the priority + reflects the prices at alert time rather than any stale stored + percentage. Missing prices, zero original prices, and price increases + resolve to priority 0. + + Args: + document: Product document supplying the prices. + + Returns: + Priority from 0 to 10, one bucket per 10 percent of discount. + """ + + current_price = document.get("currentPrice") + original_price = document.get("originalPrice") + if current_price is None or not original_price: + return 0 + + percentage = int(((original_price - current_price) / original_price) * 100) + return min(max(percentage // 10, 0), 10) + + def build_doc_id( product_id: str, category_identifier: str, use_composite: bool = False ) -> str: diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index b03e0af..974a9bf 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -351,8 +351,11 @@ async def execute_pagination( per monitor. """ - LOGGER.info("executing pagination", extra={"category": category_identifier}) total_pages = (int(total_products) + products_per_page - 1) // products_per_page + LOGGER.info( + "executing pagination", + extra={"category": category_identifier, "total": total_pages}, + ) semaphore = asyncio.Semaphore(max_concurrent_tasks) stop_event = asyncio.Event() pagination_label = "offset" if pagination_mode == "offset" else "page"