added alerts collection dsql suggested

src/monitor_utils/constants.py

@@ -18,6 +18,7 @@ MONGO_DATABASE_NAME = "tpv_woot"
 MONGO_URL = (
     "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
 )
+ALERTS_COLLECTION_NAME = "alerts"
 PRODUCTION_BOT_TOKEN = (
     "MTQwMDE4NDk2MzY4NDYzMDUyOQ.GOrqOb.6frT3upUQ5B9XubVOIT9pdzVcYVXSrnEMK7RlA"
 )
@@ -36,7 +37,6 @@ RETRY_LIMIT = 5
 RETRY_WAIT = 3
 MIN_BACKOFF_DELAY = 0.1
 MAX_BACKOFF_DELAY = 2.0
-STOP_POLL_DELAY = 5
 FAILURE_DELAY = 3
 
 TIMEOUT_SECONDS = 30

src/monitor_utils/database.py

@@ -2,7 +2,8 @@
 
 Provides the shared Motor client, collection access, and the upsert path that
 rewrites current product fields while preserving prior prices in a pastPrices
-map and stamping lastUpdated on every write.
+array deduplicated by price, stamping lastUpdated on every write, and emitting
+alert references for new products and price changes.
 """
 
 import logging
@@ -14,7 +15,11 @@ from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo import UpdateOne
 from pymongo.errors import PyMongoError
 
-from monitor_utils.constants import ProductStatus, MONGO_DATABASE_NAME, MONGO_URL
+from monitor_utils.constants import (
+    ALERTS_COLLECTION_NAME,
+    MONGO_DATABASE_NAME,
+    MONGO_URL,
+)
 from monitor_utils.models import Product
 
 LOGGER = logging.getLogger(__name__)
@@ -65,53 +70,65 @@ async def add_products_to_database(
     products: list[Product],
     collection: AsyncIOMotorCollection,
     category_identifier: str,
+    monitor_name: str,
     use_composite: bool = False,
 ):
-    """Upsert products, tracking price history and last-updated time.
+    """Upsert products, tracking deduplicated price history and emitting alerts.
 
-    Rewrites current fields on each upsert while preserving prior prices in a
+    Rewrites current fields on each upsert. Prices are recorded in a
-    pastPrices map keyed by epoch-millisecond timestamps, appending only when
+    pastPrices array of price/date entries; a price already present in the
-    the price differs from the stored value. Flags dropped prices as pending
+    array is never recorded again, so each distinct price keeps only its
-    and records the drop's prior price.
+    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.
 
     Args:
         products: Products to upsert.
         collection: Target MongoDB collection.
         category_identifier: Category suffix appended to composite ids.
+        monitor_name: Monitor name stamped onto emitted alerts for consumer
+            routing.
         use_composite: Build composite ids so multi-variant monitors (Woot)
             keep one doc per variant; other monitors collapse to one doc per
             product_id.
     """
+
     doc_ids = [
         build_doc_id(p.product_id, category_identifier, use_composite) for p in products
     ]
-    cursor = collection.find({"_id": {"$in": doc_ids}}, {"currentPrice": 1})
+    cursor = collection.find(
-    existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor}
+        {"_id": {"$in": doc_ids}}, {"currentPrice": 1, "pastPrices": 1}
+    )
+    existing = {str(doc["_id"]): doc async for doc in cursor}
 
     ops = []
+    alerts = []
     now = datetime.now(timezone.utc)
-    price_key = str(int(now.timestamp() * 1000))
     for product, doc_id in zip(products, doc_ids):
         document = convert_product_to_mongo(product)
         document["_id"] = doc_id
-        old_price = existing_prices.get(doc_id)
+        existing_doc = existing.get(doc_id)
+        old_price = existing_doc.get("currentPrice") if existing_doc else None
         new_price = document.get("currentPrice")
         price_changed = new_price is not None and new_price != old_price
         set_on_insert = {"discoveredDate": now}
-        if old_price is None:
+        if existing_doc is None or price_changed:
-            set_on_insert["status"] = ProductStatus.PENDING
+            alerts.append(build_alert(monitor_name, document, now))
-        elif price_changed:
-            document["status"] = ProductStatus.PENDING
-            document["priceBeforeDrop"] = old_price
 
         update = {"$set": document, "$setOnInsert": set_on_insert}
-        if price_changed:
+        recorded_prices = extract_recorded_prices(existing_doc)
-            update["$set"][f"pastPrices.{price_key}"] = new_price
+        if new_price is not None and new_price not in recorded_prices:
+            update["$push"] = {"pastPrices": {"price": new_price, "date": now}}
 
         ops.append(UpdateOne({"_id": doc_id}, update, upsert=True))
 
     try:
         await collection.bulk_write(ops, ordered=False)
+        if alerts:
+            await get_collection(ALERTS_COLLECTION_NAME).insert_many(
+                alerts, ordered=False
+            )
     except PyMongoError:
         LOGGER.exception(
             "failed to add products to database",
@@ -120,6 +137,54 @@ async def add_products_to_database(
         raise
 
 
+def extract_recorded_prices(existing_doc: dict | None) -> set[float]:
+    """Return the set of prices already recorded in a document's pastPrices.
+
+    Args:
+        existing_doc: Existing product document, or None for new products.
+
+    Returns:
+        Prices present in the pastPrices array, empty when none exist.
+    """
+
+    if not existing_doc:
+        return set()
+
+    return {entry["price"] for entry in existing_doc.get("pastPrices", [])}
+
+
+def build_alert(monitor_name: str, document: dict, now: datetime) -> dict:
+    """Build an alert referencing a product for consumer delivery.
+
+    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.
+        now: Detection timestamp stored as createdAt.
+
+    Returns:
+        The alert document for insertion.
+    """
+
+    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),
+        "claimed": False,
+        "attempts": 0,
+        "createdAt": now,
+    }
+
+
 def build_doc_id(
     product_id: str, category_identifier: str, use_composite: bool = False
 ) -> str:
@@ -133,6 +198,7 @@ def build_doc_id(
     Returns:
         The composite id when use_composite is set, otherwise product_id.
     """
+
     return f"{product_id}_{category_identifier}" if use_composite else product_id