Merge pull request 'added no msrp prirority so they don't starve in queue' (#10) from development into main

Reviewed-on: #10

build/lib/monitor_utils/config.py

@@ -0,0 +1,63 @@
+"""Config loading and mode helpers shared across monitors and the bot."""
+
+import logging
+from enum import Enum
+
+from monitor_utils.constants import LOGGERS
+from monitor_utils.database import get_collection
+
+CONFIGS_COLLECTION_NAME = "configs"
+
+
+class ConfigNotFoundError(Exception):
+    """Raised when a monitor has no config document in Mongo."""
+
+
+class DevelopmentMode(str, Enum):
+    """Runtime mode selecting production or development behavior."""
+
+    PRODUCTION = "PRODUCTION"
+    DEVELOPMENT = "DEVELOPMENT"
+
+
+def is_production(development_mode: DevelopmentMode) -> bool:
+    """Return True when the mode is PRODUCTION."""
+
+    return development_mode == DevelopmentMode.PRODUCTION
+
+
+async def get_config(monitor_name: str, use_production: bool = False) -> dict:
+    """Load a monitor's config document from Mongo.
+
+    Reads the configs collection in production and configs_testing in
+    development, so a dev run never reads production routing.
+
+    Args:
+        monitor_name: Monitor identifier (the config document _id).
+        use_production: Read the production configs collection instead of
+            configs_testing.
+
+    Returns:
+        The config document, including its categories and channels.
+
+    Raises:
+        ConfigNotFoundError: When no document exists for the monitor.
+    """
+
+    collection_name = (
+        CONFIGS_COLLECTION_NAME
+        if use_production
+        else f"{CONFIGS_COLLECTION_NAME}_testing"
+    )
+    config = await get_collection(collection_name).find_one({"_id": monitor_name})
+    if config is None:
+        raise ConfigNotFoundError(monitor_name)
+
+    return config
+
+
+def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING):
+    """Set the given loggers to the given level to quiet noisy libraries."""
+
+    for name in loggers:
+        logging.getLogger(name).setLevel(level)

build/lib/monitor_utils/constants.py

@@ -0,0 +1,70 @@
+"""Shared constants for monitors and the consumer bot.
+
+Holds Mongo connection settings, bot tokens, retry and timeout tuning,
+client fingerprint identifiers, webhook routing maps, product status values,
+and the expected error tuple used for known-failure handling.
+"""
+
+import asyncio
+from enum import Enum
+from json import JSONDecodeError
+
+import aiohttp
+import requests
+from aioweb import AiowebError
+from curl_cffi.requests.errors import RequestsError
+from pymongo.errors import PyMongoError
+
+MONGO_DATABASE_NAME = "tpv_woot"
+MONGO_URL = (
+    "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
+)
+LOGGING_LEVEL = "INFO"
+LOGGERS = [
+    "discord",
+    "pymongo",
+    "aiohttp",
+]
+
+RETRY_LIMIT = 5
+RETRY_WAIT = 3
+CYCLE_DELAY = 0.30
+MIN_BACKOFF_DELAY = 0.1
+MAX_BACKOFF_DELAY = 2.0
+FAILURE_DELAY = 3
+
+TIMEOUT_SECONDS = 30
+CYCLE_TIMEOUT_SECONDS = 600
+# curl_cffi's AsyncSession defaults to 10 handles and blocks on an untimed
+# pool.get() past that, so a monitor fanning out wider silently serialises with
+# no timeout covering the wait. Handles are allocated lazily (the pool holds
+# None placeholders until popped), so a generous ceiling costs a low-concurrency
+# monitor nothing. Monitors fanning out past this pass max_clients explicitly.
+MAX_CURL_HANDLES = 64
+# Budget for closing a session. curl_cffi's close awaits the multi handle with
+# no timeout of its own, so a wedged transfer would otherwise park whatever is
+# closing the session, including an `async with` exit.
+SESSION_CLOSE_TIMEOUT_SECONDS = 10
+CLIENT_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"]
+CLIENT_TIMEOUT = aiohttp.ClientTimeout(
+    total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30
+)
+EXPECTED_TASK_ERRORS = (
+    RequestsError,
+    aiohttp.ClientError,
+    asyncio.TimeoutError,
+    requests.Timeout,
+    requests.HTTPError,
+    PyMongoError,
+    JSONDecodeError,
+    AiowebError,
+)
+
+
+class ProductStatus(str, Enum):
+    """Lifecycle status for a monitored product document."""
+
+    PENDING = "pending"
+    SENT = "sent"
+    SKIPPED = "skipped"
+    FAILED = "failed"

build/lib/monitor_utils/database.py

@@ -0,0 +1,369 @@
+"""Database helpers for upserting monitored products and tracking price history.
+
+Provides the shared Motor client, collection access, and the upsert path that
+rewrites current product fields while preserving prior prices in a pastPrices
+array deduplicated by price, stamping lastUpdated on every write, and emitting
+alert references for new products and price changes.
+"""
+
+import logging
+from dataclasses import asdict
+from datetime import datetime, timezone
+from functools import cache
+
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
+from pymongo import UpdateOne
+from pymongo.errors import BulkWriteError, PyMongoError
+
+from monitor_utils.constants import MONGO_DATABASE_NAME, MONGO_URL
+from monitor_utils.models import Product
+
+LOGGER = logging.getLogger(__name__)
+
+
+def get_collection(collection_name: str) -> AsyncIOMotorCollection:
+    """Return the named collection from the shared client's database.
+
+    Args:
+        collection_name: Name of the collection to return.
+
+    Returns:
+        The requested collection.
+    """
+
+    return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
+
+
+def resolve_alerts_collection(
+    product_collection: AsyncIOMotorCollection,
+) -> AsyncIOMotorCollection:
+    """Return the alerts collection beside a product collection.
+
+    Alerts co-locate with the products they reference: the same database and
+    the same development mode. The mode is read from the product collection's
+    own name (setup_config appends a _testing suffix outside production), so a
+    monitor writing products to <name>_testing also writes alerts to
+    alerts_testing and the development bot draining alerts_testing sees them.
+
+    Args:
+        product_collection: The mode-resolved product collection the monitor
+            already writes to.
+
+    Returns:
+        The matching alerts collection in the same database and mode.
+    """
+
+    suffix = "_testing" if product_collection.name.endswith("_testing") else ""
+    return product_collection.database[f"alerts{suffix}"]
+
+
+@cache
+def get_mongo_client() -> AsyncIOMotorClient:
+    """Return a cached Motor client configured for the shared cluster.
+
+    Returns:
+        The pooled, timezone-aware Motor client.
+    """
+
+    return AsyncIOMotorClient(
+        MONGO_URL,
+        minPoolSize=10,
+        connectTimeoutMS=20_000,
+        socketTimeoutMS=60_000,
+        serverSelectionTimeoutMS=30_000,
+        retryWrites=True,
+        retryReads=True,
+        tz_aware=True,
+        tzinfo=timezone.utc,
+    )
+
+
+async def get_product_from_database(
+    collection: AsyncIOMotorCollection,
+    product_id: str,
+) -> dict:
+    """Fetch a single product document by its _id."""
+
+    return await collection.find_one({"_id": product_id})
+
+
+async def add_products_to_database(
+    products: list[Product],
+    collection: AsyncIOMotorCollection,
+    category_identifier: str,
+    monitor_name: str,
+    use_composite: bool = False,
+):
+    """Upsert products, tracking deduplicated price history and emitting alerts.
+
+    Rewrites current fields on each upsert. priceBeforeDrop is derived here,
+    not by monitors: it is set to the previous stored currentPrice when the new
+    price is lower (a drop) and cleared otherwise, so any price_before_drop on
+    the incoming product is ignored. 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 never-before-recorded prices
+    upsert an alert referencing the product; a price returning to a value
+    already in pastPrices emits nothing, so per-source price oscillation
+    (member vs non-member quotes for the same SKU) alerts once per distinct
+    price instead of every run. Alerts are deduplicated so a product holds
+    at most one unclaimed alert regardless of concurrent category or page
+    tasks detecting it, and write 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.
+    """
+
+    # Callers are expected to filter empty batches, but guard here too: an empty
+    # products list would make collection.bulk_write([]) raise InvalidOperation,
+    # which callers would then log as a spurious category failure every cycle.
+    if not products:
+        return
+
+    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, "pastPrices": 1}
+    )
+    existing = {str(doc["_id"]): doc async for doc in cursor}
+
+    ops = []
+    alert_ops = []
+    now = datetime.now(timezone.utc)
+    for product, doc_id in zip(products, doc_ids):
+        document = convert_product_to_mongo(product)
+        document["_id"] = doc_id
+        existing_doc = existing.get(doc_id)
+        new_price = document.get("currentPrice")
+        previous_price = existing_doc.get("currentPrice") if existing_doc else None
+        document["priceBeforeDrop"] = resolve_price_before_drop(
+            previous_price, new_price
+        )
+
+        recorded_prices = extract_recorded_prices(existing_doc)
+        never_seen_price = new_price is not None and new_price not in recorded_prices
+        set_on_insert = {"discoveredDate": now}
+        if existing_doc is None or never_seen_price:
+            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}
+        if never_seen_price:
+            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 alert_ops:
+            await write_alert_ops(alert_ops, resolve_alerts_collection(collection))
+    except PyMongoError:
+        LOGGER.exception(
+            "failed to add products to database",
+            extra={"category": category_identifier},
+        )
+        raise
+
+
+async def write_alert_ops(
+    alert_ops: list[UpdateOne], alerts_collection: AsyncIOMotorCollection
+):
+    """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.
+        alerts_collection: The mode-resolved alerts collection to write to,
+            supplied by the caller so the writer stays mode-agnostic.
+    """
+
+    try:
+        await alerts_collection.bulk_write(alert_ops, ordered=False)
+    except BulkWriteError as exc:
+        real_errors = [
+            write_error
+            for write_error in exc.details["writeErrors"]
+            if write_error.get("code") != 11000
+        ]
+        if real_errors:
+            raise
+
+
+def resolve_price_before_drop(
+    previous_price: float | None, new_price: float | None
+) -> float | None:
+    """Return the pre-drop price when new_price is a drop from previous_price.
+
+    Owns price-drop detection so monitors do not: a drop is a new price strictly
+    below the previously stored one, and its priceBeforeDrop is that prior price.
+    Any other case (no prior price, unchanged, or a rise) is not a drop.
+
+    Args:
+        previous_price: The product's last stored currentPrice, or None when new.
+        new_price: The incoming currentPrice.
+
+    Returns:
+        previous_price on a drop, otherwise None.
+    """
+
+    if (
+        previous_price is not None
+        and new_price is not None
+        and new_price < previous_price
+    ):
+        return previous_price
+
+    return None
+
+
+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 pricing for the
+            priority calculation.
+        now: Detection timestamp stored as createdAt.
+
+    Returns:
+        The alert document for upsert via $setOnInsert.
+    """
+
+    return {
+        "monitor": monitor_name,
+        "product_id": document["_id"],
+        "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:
+    """Return the composite id when use_composite is set, else the bare product id.
+
+    Args:
+        product_id: Retailer-specific product identifier.
+        category_identifier: Category suffix for multi-variant monitors.
+        use_composite: Append category_identifier to disambiguate variants.
+
+    Returns:
+        The composite id when use_composite is set, otherwise product_id.
+    """
+
+    return f"{product_id}_{category_identifier}" if use_composite else product_id
+
+
+def convert_product_to_mongo(product: Product) -> dict:
+    """Flatten a Product into its MongoDB document representation.
+
+    Stamps lastUpdated with the current UTC time. Prices are already floats via
+    the Product's pydantic validation (numeric strings coerced, comma strings
+    rejected at construction), so no coercion happens here. The _id is assigned
+    by the caller via build_doc_id.
+
+    Args:
+        product: Product to convert.
+
+    Returns:
+        The MongoDB document dict for the product.
+    """
+
+    data = asdict(product)
+    metadata = data["metadata"]
+    timing = data["timing"]
+    pricing = data["pricing"]
+    availability = data["availability"]
+    current_price = pricing["current_price"]
+    original_price = pricing["original_price"]
+
+    return {
+        "title": data["title"],
+        "url": data["url"],
+        "currentPrice": current_price,
+        "originalPrice": original_price,
+        "imageUrl": metadata["image_url"],
+        "customizationIndex": metadata["customization_index"],
+        "memberNonMemberComparison": metadata["member_non_member_comparison"],
+        "overstockText": metadata["overstock_text"],
+        "affiliateUrl": metadata["affiliate_url"],
+        "promoCode": metadata["promo_code"],
+        "promoPercentage": metadata["promo_percentage"],
+        "promoDollarOff": metadata["promo_dollar_off"],
+        "promoMinSpend": metadata["promo_min_spend"],
+        "startDate": timing["start_date"],
+        "endDate": timing["end_date"],
+        "priceBeforeDrop": pricing["price_before_drop"],
+        "eduPrice": pricing["edu_price"],
+        "hasMsrp": pricing["has_msrp"],
+        "available": availability["available"],
+        "availabilityLastChecked": availability["availability_last_checked"],
+        "condition": availability["condition"],
+        "appExclusive": availability["app_exclusive"],
+        "memberPick": availability["member_pick"],
+        "memberDeal": availability["member_deal"],
+        "lastUpdated": datetime.now(timezone.utc),
+    }

build/lib/monitor_utils/dsql_shortlinks.py

@@ -0,0 +1,98 @@
+# Idk who made this i was provided it by louis
+import asyncio
+from typing import Dict
+
+import httpx
+
+"""
+Minimal usage:
+
+    from dsql_shortlinks import generate_short_link_sync
+
+    result = generate_short_link_sync(
+        "https://example.com/product",
+        api_key="YOUR_DSQL_KEY",
+    )
+
+    print(result["short_url"])
+"""
+
+
+DEFAULT_BASE_URL = "https://api.profitlounge.us/link/affiliate"
+
+
+class DsqlShortlinks:
+    def __init__(
+        self,
+        api_key: str,
+        *,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout: int = 60,
+    ) -> None:
+        self.api_key = api_key
+        self.base_url = base_url
+        self.timeout = timeout
+
+    async def create_short_link(self, url: str) -> Dict[str, str]:
+        headers = {"x-api-key": self.api_key}
+        payload = {"url": url}
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    self.base_url,
+                    headers=headers,
+                    json=payload,
+                    timeout=self.timeout,
+                )
+                response.raise_for_status()
+                data = response.json()
+                if isinstance(data, dict) and data.get("short_url"):
+                    return data
+        except Exception:
+            pass
+
+        return {"short_url": url, "long_url": url}
+
+    def create_short_link_sync(self, url: str) -> Dict[str, str]:
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+
+        if running_loop and running_loop.is_running():
+            raise RuntimeError(
+                "create_short_link_sync() cannot run inside an active event loop. "
+                "Use await create_short_link(...) instead."
+            )
+
+        return asyncio.run(self.create_short_link(url))
+
+
+async def generate_short_link(
+    url: str, api_key: str, timeout: int = 60
+) -> Dict[str, str]:
+    client = DsqlShortlinks(api_key, timeout=timeout)
+    return await client.create_short_link(url)
+
+
+def generate_short_link_sync(
+    url: str, api_key: str, timeout: int = 60
+) -> Dict[str, str]:
+    client = DsqlShortlinks(api_key, timeout=timeout)
+    return client.create_short_link_sync(url)
+
+
+if __name__ == "__main__":
+    EXAMPLE_URL = "https://woot.com/"
+    EXAMPLE_API_KEY = "YOUR_DSQL_KEY"
+
+    print("Example:")
+    print("from dsql_shortlinks import generate_short_link_sync")
+    print()
+    print("result = generate_short_link_sync(")
+    print(f'    "{EXAMPLE_URL}",')
+    print(f'    api_key="{EXAMPLE_API_KEY}",')
+    print(")")
+    print()
+    print('print(result["short_url"])')

build/lib/monitor_utils/models.py

@@ -0,0 +1,163 @@
+"""Dataclass models shared across monitors and the consumer bot."""
+
+from dataclasses import field
+from datetime import datetime
+
+from pydantic.dataclasses import dataclass
+
+
+@dataclass
+class CycleResult:
+    """Tallied outcome of one monitor cycle.
+
+    Pure counts for the caller to log or act on as it sees fit; no policy is
+    baked in.
+
+    Attributes:
+        successes: Category tasks that completed without an expected error.
+        failures: Category tasks that raised an expected error.
+        skipped: True when pre_task_func failed and no category tasks ran.
+    """
+
+    successes: int = 0
+    failures: int = 0
+    skipped: bool = False
+
+
+@dataclass
+class ConsumerConfig:
+    """Configuration for a single retailer's consumer.
+
+    Attributes:
+        monitor_name: Retailer name (matches setup_config key).
+        collection: MongoDB collection for this retailer's products.
+        channel_ids: Discord webhook/channel routing config.
+        min_percentage: Minimum discount % to trigger a ping.
+        ping_threshold: Price floor for the secondary 40% rule (e.g. Woot pings
+            on 40%+ discounts if the item is >= this price). None disables
+            the secondary rule.
+    """
+
+    monitor_name: str
+    collection: object
+    channel_ids: dict
+    min_percentage: int
+    ping_threshold: int | None = None
+
+
+@dataclass
+class PricingInfo:
+    """Price fields for a product.
+
+    Attributes:
+        original_price: Pre-discount price.
+        current_price: Current listed price.
+        price_before_drop: Stored price prior to the last detected drop.
+        edu_price: Education-store price when offered.
+        past_prices: Prior price history as price/date entries, oldest first.
+    """
+
+    original_price: float
+    current_price: float
+    price_before_drop: float | None = None
+    edu_price: float | None = None
+    past_prices: list[dict] | None = None
+    has_msrp: bool = True
+
+
+@dataclass
+class AvailabilityInfo:
+    """Availability and membership flags for a product.
+
+    Attributes:
+        available: Whether the product is purchasable.
+        condition: Product condition such as new or open box.
+        app_exclusive: Only purchasable through the retailer's app.
+        member_pick: Flagged as a member pick offer.
+        member_deal: Flagged as a member deal offer.
+        availability_last_checked: When available was last verified against
+            the retailer, letting monitors skip rechecks within an interval.
+    """
+
+    available: bool | None = None
+    condition: str | None = None
+    app_exclusive: bool | None = None
+    member_pick: bool | None = None
+    member_deal: bool | None = None
+    availability_last_checked: datetime | None = None
+
+
+@dataclass
+class ProductMetadata:
+    """Presentation and routing metadata for a product.
+
+    Attributes:
+        image_url: Product image URL.
+        channel_id: Resolved Discord channel id for the ping.
+        customization_index: Config number when priced via customization.
+        member_non_member_comparison: Display string comparing member and
+            non-member pricing.
+        overstock_text: Overstock label text when present.
+        percentage: Computed discount percentage.
+        affiliate_url: Affiliate link when available.
+        promo_code: Event-level checkout code (e.g. "BIRTHDAY26") when the
+            product's event advertises one, else None.
+        promo_percentage: Advertised percent off from a percentage code
+            ("extra 22% off" -> 22); None for dollar codes or no code. This is
+            the claimed discount, not the actual price delta.
+        promo_dollar_off: Advertised dollar amount off from a spend-threshold
+            code ("get an extra $22 off" -> 22); None for percentage codes or
+            no code. Pairs with promo_min_spend.
+        promo_min_spend: Spend threshold a dollar code requires ("Spend $100"
+            -> 100); None unless promo_dollar_off is set.
+    """
+
+    image_url: str | None = None
+    channel_id: str | None = None
+    customization_index: str | None = None
+    member_non_member_comparison: str | None = None
+    overstock_text: str | None = None
+    percentage: int | None = None
+    affiliate_url: str | None = None
+    promo_code: str | None = None
+    promo_percentage: int | None = None
+    promo_dollar_off: int | None = None
+    promo_min_spend: int | None = None
+
+
+@dataclass
+class ProductTiming:
+    """Lifecycle timestamps for a product.
+
+    Attributes:
+        discovered_date: When the product was first stored.
+        start_date: Offer start when the retailer provides one.
+        end_date: Offer end when the retailer provides one.
+    """
+
+    discovered_date: datetime | None = None
+    start_date: datetime | None = None
+    end_date: datetime | None = None
+
+
+@dataclass
+class Product:
+    """A monitored product assembled from retailer data.
+
+    Attributes:
+        product_id: Retailer-specific product identifier.
+        title: Product title.
+        url: Product page URL.
+        pricing: Price fields.
+        metadata: Presentation and routing metadata.
+        timing: Lifecycle timestamps.
+        availability: Availability and membership flags.
+    """
+
+    product_id: str
+    title: str
+    url: str
+    pricing: PricingInfo
+    metadata: ProductMetadata = field(default_factory=ProductMetadata)
+    timing: ProductTiming = field(default_factory=ProductTiming)
+    availability: AvailabilityInfo = field(default_factory=AvailabilityInfo)

build/lib/monitor_utils/runner.py

@@ -0,0 +1,691 @@
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable, Iterable
+from contextlib import asynccontextmanager, nullcontext
+
+import aiohttp
+from aioproxies import AioProxies
+from curl_cffi.requests import AsyncSession
+from log_setup import setup_logging
+from motor.motor_asyncio import AsyncIOMotorCollection
+
+from monitor_utils.config import (
+    DevelopmentMode,
+    get_config,
+    is_production,
+    set_logger_levels,
+)
+from monitor_utils.constants import (
+    CYCLE_TIMEOUT_SECONDS,
+    EXPECTED_TASK_ERRORS,
+    LOGGING_LEVEL,
+    MAX_CURL_HANDLES,
+    TIMEOUT_SECONDS,
+)
+from monitor_utils.database import get_collection
+from monitor_utils.sessions import get_default_session
+from monitor_utils.models import (
+    AvailabilityInfo,
+    CycleResult,
+    PricingInfo,
+    Product,
+    ProductMetadata,
+    ProductTiming,
+)
+
+LOGGER = logging.getLogger(__name__)
+
+
+def setup_monitor_logging(monitor_name: str, logging_level=LOGGING_LEVEL):
+    """Configure JSON logging for a monitor and quiet noisy libraries.
+
+    Call once at module import, at global scope, before creating module loggers,
+    so logging is configured synchronously at startup rather than inside the
+    async init_monitor. Logs go to logs/latest.log with rotated history keyed by
+    monitor_name.
+
+    Args:
+        monitor_name: Monitor name used as the rotated-history base.
+        logging_level: Root logging level.
+    """
+
+    setup_logging(
+        name="logs/latest.log",
+        history_name=monitor_name,
+        level=logging_level,
+        output="json",
+    )
+    set_logger_levels()
+
+
+async def init_monitor(
+    monitor_name: str,
+    development_mode: DevelopmentMode,
+    production_proxy_path: str = "proxies.txt",
+    testing_proxy_path: str = "tests/configs/proxies.txt",
+) -> tuple[list[str] | dict[str, str], AsyncIOMotorCollection, AioProxies]:
+    """Load a monitor's config, collection, and proxies for the mode.
+
+    Async because config now lives in Mongo; call it from inside the monitor's
+    event loop (in main), not at module import. Logging is configured separately
+    by setup_monitor_logging at module scope and is not touched here.
+
+    Args:
+        monitor_name: Monitor identifier used for config and collection lookup.
+        development_mode: Selects production or testing config, collection,
+            and proxy path.
+        production_proxy_path: Proxy file path used in production mode.
+        testing_proxy_path: Proxy file path used in testing mode.
+
+    Returns:
+        Tuple of category names, Mongo collection, and loaded proxies.
+    """
+
+    use_production = is_production(development_mode)
+    category_names, _, collection = await setup_config(
+        monitor_name=monitor_name, development_mode=development_mode
+    )
+    proxy_path = production_proxy_path if use_production else testing_proxy_path
+    proxies = AioProxies.from_file(path=proxy_path)
+    return category_names, collection, proxies
+
+
+async def setup_config(
+    monitor_name: str, development_mode: DevelopmentMode
+) -> tuple[
+    list[str] | dict[str, str], dict[str, str | list[str]], AsyncIOMotorCollection
+]:
+    """Load monitor config and resolve the Mongo collection for the mode.
+
+    Testing mode reads the configs_testing collection and appends a _testing
+    suffix to the product collection name so production data is never touched
+    during development.
+
+    Args:
+        monitor_name: Monitor identifier used for config and collection lookup.
+        development_mode: Selects the production or testing config and
+            collection.
+
+    Returns:
+        Tuple of category names, channel id mapping, and Mongo collection.
+    """
+
+    use_production = is_production(development_mode)
+    config = await get_config(
+        monitor_name=monitor_name,
+        use_production=use_production,
+    )
+
+    category_names = config["categories"]
+    collection = get_collection(
+        monitor_name if use_production else f"{monitor_name}_testing"
+    )
+    return category_names, config["channels"], collection
+
+
+def combined_discount(base_percentage: int, promo_percentage: int | None) -> int:
+    """Combine an on-sale discount with a percentage code applied at checkout.
+
+    A "extra N% off" code discounts the already-reduced sale price, so the two
+    compound rather than add: 20% off then an extra 30% code is 44% off list,
+    not 50%. Returns base_percentage unchanged when there is no percentage code.
+
+    Args:
+        base_percentage: The on-sale discount off list, as a whole percent.
+        promo_percentage: The code's advertised percent off, or None.
+
+    Returns:
+        The combined discount off list, rounded to a whole percent.
+    """
+
+    if not promo_percentage:
+        return base_percentage
+
+    return round(100 - (100 - base_percentage) * (100 - promo_percentage) / 100)
+
+
+def build_product_details(
+    product: dict,
+    min_percentage: int,
+    min_price_drop: float | None = None,
+) -> Product | None:
+    """Build a Product from a Mongo document if it passes discount thresholds.
+
+    Skips documents with a missing or zero original price. Passing requires
+    meeting min_percentage, or min_price_drop when provided. Channel routing
+    is owned by the consumer bot; no channel id is attached here.
+
+    Args:
+        product: Mongo document for the product.
+        min_percentage: Minimum discount percentage to pass.
+        min_price_drop: Optional minimum absolute price drop to pass.
+
+    Returns:
+        The built Product, or None when skipped or below thresholds.
+    """
+
+    product_id = product["_id"]
+    current_price = product["currentPrice"]
+    original_price = product["originalPrice"]
+
+    percentage = 0
+    # Defaulted, not indexed: documents written before the field existed have no
+    # hasMsrp at all, and stale ones that drop out of a monitor's feed are never
+    # re-upserted, so they would KeyError here forever. True keeps them on the
+    # pre-existing threshold path rather than silently routing them as MSRP-less.
+    has_msrp = product.get("hasMsrp", True)
+    if has_msrp:
+        if not original_price:
+            LOGGER.warning(
+                "missing or zero original price, skipping",
+                extra={"product_id": product_id},
+            )
+            return
+
+        percentage = int(((original_price - current_price) / original_price) * 100)
+        price_drop = original_price - current_price
+        # A percentage code compounds onto the sale price, so an item can clear the
+        # threshold on the combined discount even when the base discount would not.
+        # Dollar codes do not map to a percent and never affect the threshold.
+
+        effective_percentage = combined_discount(
+            percentage, product.get("promoPercentage")
+        )
+        passes_percentage = effective_percentage >= min_percentage
+        passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
+        if not passes_percentage and not passes_price_drop:
+            return
+
+    built = Product(
+        product_id=product_id,
+        title=product["title"],
+        url=product["url"],
+        metadata=ProductMetadata(
+            image_url=product["imageUrl"],
+            customization_index=product["customizationIndex"],
+            member_non_member_comparison=product["memberNonMemberComparison"],
+            overstock_text=product["overstockText"],
+            percentage=percentage,
+            affiliate_url=product.get("affiliateUrl"),
+            promo_code=product.get("promoCode"),
+            promo_percentage=product.get("promoPercentage"),
+            promo_dollar_off=product.get("promoDollarOff"),
+            promo_min_spend=product.get("promoMinSpend"),
+        ),
+        timing=ProductTiming(
+            discovered_date=product["discoveredDate"],
+            start_date=product["startDate"],
+            end_date=product["endDate"],
+        ),
+        pricing=PricingInfo(
+            current_price=current_price,
+            original_price=original_price,
+            price_before_drop=product["priceBeforeDrop"],
+            edu_price=product["eduPrice"],
+            past_prices=product.get("pastPrices"),
+            has_msrp=has_msrp,
+        ),
+        availability=AvailabilityInfo(
+            available=product["available"],
+            condition=product["condition"],
+            app_exclusive=product["appExclusive"],
+            member_pick=product["memberPick"],
+            member_deal=product["memberDeal"],
+        ),
+    )
+    return built
+
+
+async def run_monitor_cycle(
+    category_identifiers: list[str] | dict[str, str],
+    task_func: Callable[..., Awaitable[None]],
+    session: aiohttp.ClientSession | AsyncSession,
+    collection: AsyncIOMotorCollection,
+    *,
+    pre_task_func: Callable[..., Awaitable[None]] | None = None,
+    concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
+    **kwargs,
+) -> CycleResult:
+    """Run one monitor cycle over a caller-supplied session and report the outcome.
+
+    Awaits pre_task_func if given, dispatches task_func per category
+    concurrently, logs expected failures, and raises unexpected ones. The
+    caller owns the loop, the session lifetime, and the inter-cycle delay: wrap
+    this in a `while True` that opens a fresh session per cycle and sleeps
+    between cycles — or use run_bounded_cycle, which owns the session and a
+    hard cycle budget and leaves only the loop and the sleep to the caller.
+    The returned CycleResult is pure counts for the caller to log or act on;
+    this function applies no policy to it.
+
+    session and collection are the two per-cycle resources, passed to every
+    callback (pre_task_func, concurrent_pre_task_func, and each task_func) just
+    like each other. A callback that does not need one still receives it, so
+    every callback must accept **kwargs to absorb the ones it ignores.
+
+    Args:
+        category_identifiers: List of category ids, or a mapping of category
+            id to query id; mapping entries pass query_category_id to
+            task_func.
+        task_func: Awaitable invoked per category with the identifier, session,
+            collection, and **kwargs.
+        session: Open HTTP session for this cycle, built and closed by the
+            caller. Proxies are the caller's responsibility, bound to the
+            session (see get_default_session).
+        collection: Mongo collection the tasks persist products to; forwarded to
+            every callback so any that writes has it.
+        pre_task_func: Optional awaitable run once at the start of the cycle,
+            awaited before any category tasks are dispatched and receiving
+            session, collection, and **kwargs. Use it for setup the category
+            tasks depend on (session headers, cookies, fetched state); if it
+            raises an expected task error the cycle is skipped and a skipped
+            CycleResult returned.
+        concurrent_pre_task_func: Optional awaitable run once per cycle, gathered
+            alongside the category tasks and receiving session, collection, and
+            **kwargs. Category tasks must not depend on its side effects
+            (session headers, cookies, fetched state) since no ordering is
+            guaranteed.
+        **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
+            and every task_func call.
+
+    Returns:
+        A CycleResult tallying task successes and expected-error failures, or a
+        skipped result when pre_task_func failed.
+
+    Raises:
+        Exception: Any unexpected (non-EXPECTED_TASK_ERRORS) error raised by a
+            task or pre_task_func propagates, failing loud so the caller's loop
+            crashes rather than silently retrying a broken monitor.
+    """
+
+    if pre_task_func:
+        try:
+            await pre_task_func(session=session, collection=collection, **kwargs)
+        except EXPECTED_TASK_ERRORS as error:
+            LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
+            return CycleResult(skipped=True)
+
+    LOGGER.info("starting tasks")
+    if isinstance(category_identifiers, dict):
+        tasks = [
+            task_func(
+                category_id, session, collection, query_category_id=query, **kwargs
+            )
+            for category_id, query in category_identifiers.items()
+        ]
+        category_keys = list(category_identifiers.keys())
+    else:
+        tasks = [
+            task_func(category_identifier, session, collection, **kwargs)
+            for category_identifier in category_identifiers
+        ]
+        category_keys = list(category_identifiers)
+
+    if concurrent_pre_task_func:
+        tasks.append(
+            concurrent_pre_task_func(session=session, collection=collection, **kwargs)
+        )
+        category_keys.append("pre_task")
+
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    successes = 0
+    failures = 0
+    for category_identifier, result in zip(category_keys, results):
+        if isinstance(result, EXPECTED_TASK_ERRORS):
+            LOGGER.warning(
+                "failed to get products",
+                extra={"category": category_identifier},
+                exc_info=result,
+            )
+            failures += 1
+        elif isinstance(result, Exception):
+            LOGGER.error(
+                "unexpected error",
+                extra={"category": category_identifier},
+                exc_info=result,
+            )
+            raise result
+        else:
+            successes += 1
+
+    return CycleResult(successes=successes, failures=failures)
+
+
+async def run_bounded_cycle(
+    category_identifiers: list[str] | dict[str, str],
+    task_func: Callable[..., Awaitable[None]],
+    proxies: AioProxies,
+    collection: AsyncIOMotorCollection,
+    *,
+    headers: dict | None = None,
+    session_timeout: int = TIMEOUT_SECONDS,
+    max_clients: int = MAX_CURL_HANDLES,
+    cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
+    **kwargs,
+) -> CycleResult | None:
+    """Run one monitor cycle on a fresh session under a hard time budget.
+
+    The session is opened and closed inside the call, so curl multi-handle
+    state cannot outlive one cycle: a wedged transfer dies with its session
+    instead of parking the process. The budget turns an engine-level lost
+    wakeup into a caught TimeoutError and an abandoned cycle, never a
+    permanent wedge; the session's own close is bounded by
+    get_default_session, so the exit cannot park past the budget either.
+
+    The caller still owns the loop and the inter-cycle delay:
+
+        while True:
+            await run_bounded_cycle(ids, task_func, proxies, collection)
+            await asyncio.sleep(CYCLE_DELAY)
+
+    Args:
+        category_identifiers: Categories, passed through to run_monitor_cycle.
+        task_func: Awaitable invoked per category.
+        proxies: Pool bound to each cycle's fresh session.
+        collection: Mongo collection forwarded to every callback.
+        headers: Monitor-wide static headers merged into every request.
+        session_timeout: Per-request timeout on the cycle's session.
+        max_clients: Curl handle pool size. Keep at or above the monitor's
+            whole fan-out; see get_default_session for why.
+        cycle_timeout: Hard budget in seconds for the whole cycle.
+        **kwargs: Forwarded to run_monitor_cycle (pre_task_func,
+            concurrent_pre_task_func, and per-task kwargs).
+
+    Returns:
+        The cycle's CycleResult, or None when the budget expired and the
+        cycle was abandoned.
+    """
+
+    async with get_default_session(
+        proxies, headers, timeout=session_timeout, max_clients=max_clients
+    ) as session:
+        try:
+            async with asyncio.timeout(cycle_timeout):
+                return await run_monitor_cycle(
+                    category_identifiers,
+                    task_func,
+                    session=session,
+                    collection=collection,
+                    **kwargs,
+                )
+        except TimeoutError:
+            LOGGER.error(
+                "cycle exceeded budget, abandoning cycle",
+                extra={"timeout": cycle_timeout},
+            )
+            return None
+
+
+async def execute_pagination(
+    category_identifier: str,
+    total_products: int,
+    products_per_page: int,
+    func: Callable[..., Awaitable[None]],
+    session: aiohttp.ClientSession | AsyncSession,
+    *,
+    semaphore: asyncio.Semaphore | None = None,
+    start_offset: int = 0,
+    pagination_mode: str = "offset",
+    **kwargs,
+):
+    """Paginate a category by dispatching concurrent page or offset tasks to func.
+
+    Runs pages 2..N under a semaphore, retries failed pages with exponential
+    backoff, and logs any pages that still fail after retries. Page 1 is
+    assumed to have been fetched by the caller.
+
+    Args:
+        category_identifier: Label used for logging only; not passed to func.
+        total_products: Total product count used to derive page count.
+        products_per_page: Page size used to derive page count and offsets.
+        func: Awaitable task invoked per page; receives session plus either
+            offset or page_number depending on pagination_mode.
+        session: HTTP session forwarded to each func call.
+        semaphore: Optional bound on concurrent page fetches. Omit it to let a
+            monitor with few pages fan out unbounded; pass one to cap monitors
+            with many pages. Pass a single shared semaphore across all categories
+            so the limit caps total fan-out; a per-call semaphore would instead
+            bound each category in isolation, making the effective limit this
+            value times the category count.
+        start_offset: Additive base applied to every offset, for APIs whose
+            offset space does not begin at zero. Page N requests offset
+            start_offset + (N - 1) * products_per_page.
+        pagination_mode: Either offset or page, selecting the kwarg sent to func.
+        **kwargs: Forwarded verbatim to every func call. Any identifier func
+            needs (category_id, category_name, category_path, or bestbuy's
+            query_category_id) must be supplied here, since naming varies
+            per monitor.
+    """
+
+    total_pages = (int(total_products) + products_per_page - 1) // products_per_page
+    LOGGER.info(
+        "executing pagination",
+        extra={"category": category_identifier, "total": total_pages},
+    )
+    stop_event = asyncio.Event()
+    pagination_label = "offset" if pagination_mode == "offset" else "page"
+    page_range = range(2, total_pages + 1)
+
+    async def fetch_page(page_index: int):
+        """Fetch a single page or offset, optionally under the shared semaphore."""
+        async with semaphore or nullcontext():
+            if stop_event.is_set():
+                return
+
+            if pagination_mode == "offset":
+                current_offset = start_offset + (page_index - 1) * products_per_page
+                await func(
+                    session=session,
+                    offset=current_offset,
+                    **kwargs,
+                )
+            else:
+                await func(
+                    session=session,
+                    page_number=page_index,
+                    **kwargs,
+                )
+
+    failed_pages = await fetch_all_pages(
+        page_range, fetch_page, category_identifier, pagination_label
+    )
+    if failed_pages and not stop_event.is_set():
+        failed_pages = await retry_failed_pages(
+            failed_pages, fetch_page, category_identifier, pagination_label
+        )
+
+    if failed_pages and not stop_event.is_set():
+        LOGGER.error(
+            "failed to get products after retries",
+            extra={
+                "category": category_identifier,
+                "label": pagination_label,
+                "failed": failed_pages,
+            },
+        )
+
+    if stop_event.is_set():
+        LOGGER.info("pagination stopped early", extra={"category": category_identifier})
+
+
+async def fetch_all_pages(
+    page_range: Iterable[int],
+    fetch_func: Callable[[int], Awaitable[None]],
+    category_identifier: str,
+    pagination_label: str,
+) -> list[int]:
+    """Dispatch all page fetches concurrently and return failed page numbers.
+
+    Args:
+        page_range: Page numbers to fetch.
+        fetch_func: Awaitable invoked per page number.
+        category_identifier: Label used for logging.
+        pagination_label: Either offset or page, used for logging.
+
+    Returns:
+        Page numbers whose fetch raised an expected error.
+    """
+
+    tasks = [fetch_func(page) for page in page_range]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    return handle_page_results(
+        list(page_range), results, category_identifier, pagination_label
+    )
+
+
+def handle_page_results(
+    page_numbers: list[int],
+    results: list,
+    category_identifier: str,
+    pagination_label: str,
+    retry_attempt: int | None = None,
+) -> list[int]:
+    """Log page results and collect failures, raising on unexpected errors.
+
+    Expected errors are logged as warnings and their pages collected for
+    retry. Unexpected exceptions are logged and re-raised. Successful pages
+    are logged only when part of a retry attempt.
+
+    Args:
+        page_numbers: Page numbers matching results by position.
+        results: Gathered results or exceptions per page.
+        category_identifier: Label used for logging.
+        pagination_label: Either offset or page, used for logging.
+        retry_attempt: Retry attempt number when handling retry results.
+
+    Returns:
+        Page numbers whose fetch raised an expected error.
+    """
+
+    failed_pages = []
+    for page_number, result in zip(page_numbers, results):
+        if isinstance(result, EXPECTED_TASK_ERRORS):
+            LOGGER.warning(
+                "failed",
+                extra={
+                    "category": category_identifier,
+                    "label": pagination_label,
+                    "page": page_number,
+                    "retry": retry_attempt,
+                },
+                exc_info=result,
+            )
+            failed_pages.append(page_number)
+        elif isinstance(result, Exception):
+            LOGGER.error(
+                "unexpected error",
+                extra={
+                    "category": category_identifier,
+                    "label": pagination_label,
+                    "page": page_number,
+                    "retry": retry_attempt,
+                },
+                exc_info=result,
+            )
+            raise result
+        elif retry_attempt:
+            LOGGER.info(
+                "retry succeeded",
+                extra={
+                    "category": category_identifier,
+                    "label": pagination_label,
+                    "page": page_number,
+                    "retry": retry_attempt,
+                },
+            )
+
+    return failed_pages
+
+
+async def retry_failed_pages(
+    failed_pages: list[int],
+    fetch_func: Callable[[int], Awaitable[None]],
+    category_identifier: str,
+    pagination_label: str,
+    max_retries: int = 3,
+) -> list[int]:
+    """Retry failed pages with exponential backoff until exhausted or clear.
+
+    Sleeps 2**attempt seconds before each round and re-dispatches only the
+    pages that failed the previous round.
+
+    Args:
+        failed_pages: Page numbers to retry.
+        fetch_func: Awaitable invoked per page number.
+        category_identifier: Label used for logging.
+        pagination_label: Either offset or page, used for logging.
+        max_retries: Maximum retry rounds.
+
+    Returns:
+        Page numbers still failing after all retries.
+    """
+
+    retry_attempt = 1
+    while failed_pages and retry_attempt <= max_retries:
+        await asyncio.sleep(2**retry_attempt)
+        LOGGER.info(
+            "retrying failed pages",
+            extra={
+                "category": category_identifier,
+                "label": pagination_label,
+                "count": len(failed_pages),
+                "attempt": retry_attempt,
+                "max_retries": max_retries,
+            },
+        )
+        tasks = [fetch_func(page_number) for page_number in failed_pages]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        failed_pages = handle_page_results(
+            failed_pages, results, category_identifier, pagination_label, retry_attempt
+        )
+        retry_attempt += 1
+
+    return failed_pages
+
+
+def first_error(group_exc: BaseExceptionGroup) -> BaseException:
+    """Return a leaf exception of a possibly nested group, preferring an
+    unexpected one so a real bug is never masked by a routine network error."""
+    _, unexpected = group_exc.split(EXPECTED_TASK_ERRORS)
+    if unexpected is not None:
+        group_exc = unexpected
+
+    error = group_exc.exceptions[0]
+    while isinstance(error, BaseExceptionGroup):
+        error = error.exceptions[0]
+
+    return error
+
+
+@asynccontextmanager
+async def fan_out_group():
+    """A TaskGroup for page-one fan-out whose failure surfaces unwrapped.
+
+    Use it where one page's processing must be bound to the pagination that
+    page spawns: create the processing task on the yielded group, then await
+    execute_pagination in the body. The group cancels whichever sibling is
+    still running when the other fails, so a dead category stops doing work
+    nobody will use, and a cancelled cycle takes both down together.
+
+    The unwrapping is the reason this is shared rather than inlined. A bare
+    TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies
+    task results with isinstance against EXPECTED_TASK_ERRORS -- an
+    ExceptionGroup matches neither branch as intended, so a routine page
+    failure would be reclassified as an unexpected one and kill the monitor.
+    first_error unwraps to a leaf, preferring an unexpected exception so a real
+    bug is never masked by a network error.
+
+    Note the caller's own body is inside the group, so a plain exception raised
+    there (not from a child) propagates unchanged rather than being wrapped,
+    and an early `return` still awaits the child tasks.
+
+    Example:
+        async with fan_out_group() as group:
+            group.create_task(process_page(data, ...))
+            await execute_pagination(...)
+    """
+
+    try:
+        async with asyncio.TaskGroup() as group:
+            yield group
+    except BaseExceptionGroup as group_exc:
+        raise first_error(group_exc) from None

build/lib/monitor_utils/sessions.py

@@ -0,0 +1,173 @@
+"""Shared HTTP session helpers for all monitors.
+
+A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
+session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
+request_with_retries, adding per-request proxy rotation and old-style JSON/text
+return values so monitor call sites change as little as possible.
+"""
+
+import asyncio
+import logging
+import random
+
+from aioproxies import AioProxies
+from aioweb import Response
+from aioweb_tls import CurlCffi, TLSSession
+from monitor_utils.constants import (
+    CLIENT_IDENTIFIERS,
+    MAX_CURL_HANDLES,
+    RETRY_LIMIT,
+    SESSION_CLOSE_TIMEOUT_SECONDS,
+    TIMEOUT_SECONDS,
+)
+
+LOGGER = logging.getLogger(__name__)
+
+
+class BoundedCloseSession(TLSSession):
+    """A TLSSession whose close cannot outlive SESSION_CLOSE_TIMEOUT_SECONDS.
+
+    curl_cffi's close awaits the multi handle with no timeout of its own, so a
+    wedged transfer parks whatever is closing the session. A caller-side budget
+    cannot cover this reliably: asyncio.timeout cancels its task once, so a
+    cycle budget that has already fired leaves the close running as a fresh,
+    uncancelled await, and the process parks in the `async with` exit that was
+    meant to reclaim it.
+
+    Bounding it here keeps that guarantee attached to the session itself, so
+    every `async with get_default_session(...)` gets it without the call site
+    having to arrange anything. Abandoning a session leaks its handle until the
+    process restarts, which is strictly better than a wedged monitor.
+
+    Known inconsistency, deliberate: an abandoned session reports itself closed
+    while its curl handle is still open. TLSSession.close sets _closed before
+    awaiting the backend, so _is_closed() returns True on this path, and since
+    aioweb's __del__ only warns when _is_closed() is False, its leak warning is
+    suppressed for exactly the sessions that leaked. The "session close exceeded
+    budget" error logged below is therefore the only signal that it happened —
+    alert on it rather than on the library's warning.
+
+    Correcting the flag would mean overriding _is_closed too, and aioweb_tls is
+    a third-party library: this class deliberately overrides one documented
+    lifecycle method and nothing private, so an upstream release has one narrow
+    way to break it. A stale flag on a session discarded the same instant is the
+    cheaper side of that trade.
+    """
+
+    async def close(self) -> None:
+        try:
+            async with asyncio.timeout(SESSION_CLOSE_TIMEOUT_SECONDS):
+                await super().close()
+        except TimeoutError:
+            LOGGER.error(
+                "session close exceeded budget, abandoning session",
+                extra={"timeout": SESSION_CLOSE_TIMEOUT_SECONDS},
+            )
+
+
+def get_default_session(
+    proxies: AioProxies,
+    headers: dict | None = None,
+    timeout: int = TIMEOUT_SECONDS,
+    max_clients: int = MAX_CURL_HANDLES,
+) -> TLSSession:
+    """Return a TLSSession impersonating a random client fingerprint.
+
+    Fingerprint rotation lives here so every monitor rotates the same way; the
+    caller owns the session lifetime via ``async with``.
+
+    Cookies are not set here: aioweb has no constructor cookie path, and
+    monitors usually obtain them at runtime, so call session.set_cookie(name,
+    value, domain=...) where they are fetched (e.g. a pre_task_func) instead.
+
+    Args:
+        proxies: Pool attached to the session as proxy_pool, which send_request
+            rotates through per request. Required so a session can never send
+            un-proxied.
+        headers: Default headers merged into every request by aioweb. Use for
+            monitor-wide static headers; per-request headers still go through
+            send_request. The curl_cffi impersonation already supplies a
+            matching User-Agent, so one is not defaulted here.
+        timeout: Total request timeout in seconds, defaulting to TIMEOUT_SECONDS.
+            aioweb's own default is only 10s, so this restores the wider budget
+            the old sessions used and slow retailers depend on.
+        max_clients: Size of the curl handle pool, defaulting to
+            MAX_CURL_HANDLES. This is a hard concurrency ceiling: a request
+            finding the pool empty waits in pop_curl's untimed pool.get(), and
+            `timeout` does not cover that wait, only the transfer after
+            checkout. Keep it at or above the monitor's real fan-out (the sum
+            of its pagination semaphores plus any unsemaphored per-cycle
+            requests) so the wait is never reachable.
+
+    Returns:
+        A BoundedCloseSession (a TLSSession whose close is bounded) over a
+        curl_cffi backend with a random impersonation, with the proxy pool
+        attached as proxy_pool.
+    """
+    impersonate = random.choice(CLIENT_IDENTIFIERS)
+    session = BoundedCloseSession(
+        backend=CurlCffi(impersonate=impersonate),
+        headers=headers,
+        timeout=timeout,
+        max_clients=max_clients,
+    )
+    session.proxy_pool = proxies
+    return session
+
+
+async def send_request(
+    session: TLSSession,
+    url: str,
+    method: str,
+    *,
+    params: dict | None = None,
+    json_data: dict | None = None,
+    data: dict | None = None,
+    headers: dict | None = None,
+) -> Response:
+    """Send a request through a TLSSession with retries and proxy rotation.
+
+    Wraps aioweb's request_with_retries, which retries transport errors,
+    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
+    exponential backoff. The next proxy is drawn from the pool per call and
+    applied to both schemes. A non-2xx response or a fully failed request raises
+    AiowebError, an expected task error, so a failed fetch surfaces the same way
+    the old HTTPError path did. On success the aioweb Response is returned for
+    the caller to read via response.json(), response.text(), etc.
+
+    Args:
+        session: The TLSSession to send through.
+        url: Request URL.
+        method: HTTP method.
+        params: Query parameters.
+        json_data: JSON body. A dict here is sent as JSON with the
+            application/json content-type set by aioweb.
+        data: Raw request body, passed through unchanged (e.g. a url-encoded
+            form string). The caller owns the matching content-type via headers.
+            A dict is JSON-encoded by aioweb, so use json_data for JSON bodies.
+        headers: Per-request headers merged over the session defaults.
+
+    Returns:
+        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
+        response.json() (dict/list, or None if not valid JSON) or
+        response.text().
+
+    Raises:
+        AiowebError: On a non-2xx response or after all retries fail.
+    """
+
+    proxy_url = session.proxy_pool.next().url()
+    proxy = {"http": proxy_url, "https": proxy_url}
+
+    body = json_data if json_data is not None else data
+    response = await session.request_with_retries(
+        method.upper(),
+        url,
+        params=params,
+        data=body,
+        headers=headers,
+        proxies=proxy,
+        attempts=RETRY_LIMIT,
+    )
+    response.raise_for_status()
+    return response

src/monitor_utils/database.py

@@ -20,6 +20,11 @@ from monitor_utils.models import Product
 
 LOGGER = logging.getLogger(__name__)
 
+# Drain rank for products with no MSRP. Their percentage is 0 by construction, so
+# without this they sort dead last and starve behind any backlog. 6 puts them
+# directly under the 70%+ band, tying with 60-69% and broken by createdAt.
+NO_MSRP_PRIORITY = 6
+
 
 def get_collection(collection_name: str) -> AsyncIOMotorCollection:
     """Return the named collection from the shared client's database.
@@ -282,7 +287,9 @@ def calculate_priority(document: dict) -> int:
     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.
+    resolve to priority 0. A product with no MSRP has no percentage to rank
+    on and takes NO_MSRP_PRIORITY instead, since a real 0 would leave it
+    permanently behind every other alert in the claim query's sort.
 
     Args:
         document: Product document supplying the prices.
@@ -296,6 +303,9 @@ def calculate_priority(document: dict) -> int:
     if current_price is None or not original_price:
         return 0
 
+    if not document.get("hasMsrp", True):
+        return NO_MSRP_PRIORITY
+
     percentage = int(((original_price - current_price) / original_price) * 100)
     return min(max(percentage // 10, 0), 10)
 

tests/test_alert_priority.py

@@ -0,0 +1,46 @@
+"""unit test: calculate_priority ranks MSRP-less products instead of burying them
+
+run from the repo root:
+    uv run --with pytest pytest tests/test_alert_priority.py
+
+pure and offline — passes plain dicts shaped like product documents; touches no
+Mongo. The claim query sorts priority descending, so a priority of 0 means an
+alert is only drained once everything else is.
+"""
+
+from monitor_utils.database import NO_MSRP_PRIORITY, calculate_priority
+
+
+def document(current, original, **overrides) -> dict:
+    doc = {"_id": "p1", "currentPrice": current, "originalPrice": original}
+    doc.update(overrides)
+    return doc
+
+
+def test_no_msrp_takes_its_own_rank():
+    assert calculate_priority(document(50.0, 50.0, hasMsrp=False)) == NO_MSRP_PRIORITY
+
+
+def test_no_msrp_rank_sits_directly_under_the_seventy_band():
+    """It must outrank every sub-70% product and lose to every 70%+ one."""
+    assert calculate_priority(document(25.0, 100.0)) == 7  # 75% off
+    assert NO_MSRP_PRIORITY < 7
+    assert NO_MSRP_PRIORITY > calculate_priority(document(45.0, 100.0))  # 55% off
+
+
+def test_msrp_products_are_unaffected():
+    assert calculate_priority(document(25.0, 100.0, hasMsrp=True)) == 7
+    assert calculate_priority(document(50.0, 100.0, hasMsrp=True)) == 5
+
+
+def test_documents_without_the_field_keep_the_percentage_path():
+    """Documents written before hasMsrp existed must not all jump the queue."""
+    assert calculate_priority(document(50.0, 100.0)) == 5
+    assert calculate_priority(document(100.0, 100.0)) == 0
+
+
+def test_broken_prices_still_outrank_nothing():
+    """A missing or zero original price is a data defect, not a missing MSRP."""
+    assert calculate_priority(document(50.0, None, hasMsrp=False)) == 0
+    assert calculate_priority(document(50.0, 0, hasMsrp=False)) == 0
+    assert calculate_priority(document(None, 100.0, hasMsrp=False)) == 0