refactor(promo): streamline discount consumption logic

Simplify the discount consumption process in the PromoCodeService by removing unnecessary checks for active discounts and improving logging for various scenarios. Ensure that successful payments are always accounted for, even if the associated reservation has expired. This enhances clarity and reliability in the discount application workflow.

bot/services/promo_code_service.py

@@ -378,8 +378,11 @@ class PromoCodeService:
         payment_id: int
     ) -> bool:
         """
-        Consume active discount: link reservation to payment and clear active discount.
-        Call this AFTER successful payment.
+        Consume discount after successful payment.
+
+        The payment record is the source of truth. Even if the active reservation was
+        concurrently expired/cleared, we still record promo activation and reconcile
+        current_activations so successful discounted payments are always accounted for.
         """
         payment_record = await payment_dal.get_payment_by_db_id(session, payment_id)
         if not payment_record:
@@ -402,49 +405,11 @@ class PromoCodeService:
             )
             return False
 
-        active_discount = await active_discount_dal.get_active_discount(
-            session,
-            user_id,
-            include_expired=True,
-        )
-        now_utc = datetime.now(timezone.utc)
-
-        if not active_discount:
-            logging.info(
-                "Discount reservation missing at consumption time (user=%s, promo=%s, payment=%s)",
-                user_id,
-                promo_code_id,
-                payment_id,
-            )
-            return False
-
-        if active_discount.promo_code_id != promo_code_id:
-            logging.info(
-                "Active discount promo %s differs from payment promo %s; skipping consumption.",
-                active_discount.promo_code_id,
-                promo_code_id,
-            )
-            return False
-
-        if active_discount.expires_at <= now_utc:
-            logging.info(
-                "Discount reservation expired before payment consumption (user=%s, promo=%s)",
-                user_id,
-                promo_code_id,
-            )
-            cleared = await active_discount_dal.clear_active_discount_if_matches(
-                session,
-                user_id=user_id,
-                promo_code_id=promo_code_id,
-                expires_at_lte=now_utc,
-            )
-            if cleared:
-                await promo_code_dal.decrement_promo_code_usage(session, promo_code_id)
-            return False
-
         existing_activation = await promo_code_dal.get_user_activation_for_promo(
             session, promo_code_id, user_id
         )
+
+        activation_created = False
         if existing_activation:
             if existing_activation.payment_id is None:
                 updated_payment = await promo_code_dal.set_activation_payment_id(
@@ -471,13 +436,44 @@ class PromoCodeService:
                     promo_code_id,
                 )
                 return False
+            activation_created = True
 
-        await active_discount_dal.clear_active_discount_if_matches(
+        active_discount = await active_discount_dal.get_active_discount(
             session,
-            user_id=user_id,
-            promo_code_id=promo_code_id,
+            user_id,
+            include_expired=True,
         )
 
+        # Reservation is best-effort cleanup at this point; payment success already happened.
+        if active_discount and active_discount.promo_code_id == promo_code_id:
+            await active_discount_dal.clear_active_discount_if_matches(
+                session,
+                user_id=user_id,
+                promo_code_id=promo_code_id,
+            )
+        elif active_discount and active_discount.promo_code_id != promo_code_id:
+            logging.info(
+                "Active discount promo %s differs from payment promo %s during consumption.",
+                active_discount.promo_code_id,
+                promo_code_id,
+            )
+        else:
+            logging.info(
+                "Discount reservation already absent at consumption time (user=%s, promo=%s, payment=%s)",
+                user_id,
+                promo_code_id,
+                payment_id,
+            )
+
+        # If reservation was already expired/removed and we had to create activation now,
+        # restore current_activations to match the successful payment.
+        if activation_created:
+            await promo_code_dal.increment_promo_code_usage(
+                session,
+                promo_code_id,
+                allow_overflow=True,
+            )
+
         await session.flush()
         logging.info(
             "Discount consumed for user %s, promo %s, payment %s",

bot/services/stars_service.py

@@ -58,17 +58,28 @@ class StarsService:
     async def create_invoice(self, session: AsyncSession, user_id: int, months: float,
                              stars_price: int, description: str, sale_mode: str = "subscription",
                              promo_code_service=None) -> Optional[int]:
-        # Always resolve base price server-side to avoid trusting callback payload.
+        # Always resolve base price server-side and reject unknown packages.
         resolved_base_price = self._resolve_base_stars_price(months, sale_mode)
         if resolved_base_price is None:
             logging.warning(
-                "Stars base price not found for sale_mode=%s months=%s, fallback to callback price.",
+                "Stars invoice rejected: base price not found for sale_mode=%s months=%s.",
+                sale_mode,
+                months,
+            )
+            return None
+
+        original_stars_price = int(resolved_base_price)
+
+        # Detect callback tampering (or stale callback payload) and prefer server-side price.
+        if int(stars_price) != original_stars_price:
+            logging.warning(
+                "Stars callback price mismatch for user %s: callback=%s, resolved=%s, sale_mode=%s, months=%s",
+                user_id,
+                stars_price,
+                original_stars_price,
                 sale_mode,
                 months,
             )
-            original_stars_price = stars_price
-        else:
-            original_stars_price = int(resolved_base_price)
 
         # Invoice amount starts from the base price and discount is applied once.
         stars_price = original_stars_price

db/alembic_runner.py

@@ -257,6 +257,12 @@ async def run_alembic_migrations(settings: Settings, async_engine: AsyncEngine)
     """Apply Alembic migrations with bootstrap for existing installations."""
 
     alembic_config = _build_alembic_config(settings)
+    allow_bootstrap_without_legacy = os.getenv("ALEMBIC_ALLOW_STAMP_WITHOUT_LEGACY", "").lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
 
     async with async_engine.begin() as async_connection:
         (
@@ -268,17 +274,24 @@ async def run_alembic_migrations(settings: Settings, async_engine: AsyncEngine)
         )
 
         if not has_alembic_version and has_users_table:
-            if not has_legacy_migrator_table:
+            if not has_legacy_migrator_table and not allow_bootstrap_without_legacy:
                 raise RuntimeError(
                     "Alembic bootstrap refused: found existing users table without "
                     "alembic_version and without legacy schema_migrations marker. "
-                    "Cannot safely determine migration baseline."
+                    "Set ALEMBIC_ALLOW_STAMP_WITHOUT_LEGACY=true to explicitly allow "
+                    f"stamping baseline {_BASELINE_REVISION} after manual verification."
                 )
 
-            logging.info(
-                "Alembic: applying legacy migrator compatibility fixes before stamp."
-            )
-            await async_connection.run_sync(_run_legacy_migrator_compatibility)
+            if has_legacy_migrator_table:
+                logging.info(
+                    "Alembic: applying legacy migrator compatibility fixes before stamp."
+                )
+                await async_connection.run_sync(_run_legacy_migrator_compatibility)
+            else:
+                logging.warning(
+                    "Alembic: existing users table without legacy schema_migrations; "
+                    "proceeding with explicit bootstrap override flag."
+                )
 
             logging.info(
                 "Alembic: existing schema detected without alembic_version; stamping %s.",