added alerts collection dsql suggested

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
+    first-seen timestamp. New products and price changes upsert an alert
-    insert an alert referencing the product; alerts insert only after the
+    referencing the product, deduplicated so a product holds at most one
-    product write succeeds so a failed upsert cannot produce an orphaned
+    unclaimed alert regardless of concurrent category or page tasks
-    alert.
+    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:
+        if alert_ops:
-            await get_collection(ALERTS_COLLECTION_NAME).insert_many(
+            await write_alert_ops(alert_ops)
-                alerts, ordered=False
-            )
     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:

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"