diff --git a/alembic/versions/0003_promo_current_activations_not_null.py b/alembic/versions/0003_promo_current_activations_not_null.py new file mode 100644 index 0000000..cab615b --- /dev/null +++ b/alembic/versions/0003_promo_current_activations_not_null.py @@ -0,0 +1,91 @@ +"""harden promo current activations + +Revision ID: 0003_promo_current_activations_not_null +Revises: 0002_active_discount_expires_at +Create Date: 2026-02-11 00:00:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op, context +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "0003_promo_current_activations_not_null" +down_revision: Union[str, Sequence[str], None] = "0002_active_discount_expires_at" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + if context.is_offline_mode(): + op.execute( + sa.text( + "UPDATE promo_codes SET current_activations = 0 " + "WHERE current_activations IS NULL" + ) + ) + op.alter_column( + "promo_codes", + "current_activations", + existing_type=sa.Integer(), + nullable=False, + server_default=sa.text("0"), + ) + return + + bind = op.get_bind() + inspector = sa.inspect(bind) + if not inspector.has_table("promo_codes"): + return + + promo_columns = {column["name"] for column in inspector.get_columns("promo_codes")} + if "current_activations" not in promo_columns: + op.add_column( + "promo_codes", + sa.Column("current_activations", sa.Integer(), nullable=False, server_default=sa.text("0")), + ) + return + + op.execute( + sa.text( + "UPDATE promo_codes SET current_activations = 0 " + "WHERE current_activations IS NULL" + ) + ) + op.alter_column( + "promo_codes", + "current_activations", + existing_type=sa.Integer(), + nullable=False, + server_default=sa.text("0"), + ) + + +def downgrade() -> None: + if context.is_offline_mode(): + op.alter_column( + "promo_codes", + "current_activations", + existing_type=sa.Integer(), + nullable=True, + server_default=None, + ) + return + + bind = op.get_bind() + inspector = sa.inspect(bind) + if not inspector.has_table("promo_codes"): + return + + promo_columns = {column["name"] for column in inspector.get_columns("promo_codes")} + if "current_activations" in promo_columns: + op.alter_column( + "promo_codes", + "current_activations", + existing_type=sa.Integer(), + nullable=True, + server_default=None, + ) diff --git a/bot/handlers/user/payment.py b/bot/handlers/user/payment.py index c9f0e1c..a3644aa 100644 --- a/bot/handlers/user/payment.py +++ b/bot/handlers/user/payment.py @@ -1,6 +1,5 @@ import logging import json -import asyncio from datetime import datetime, timezone, timedelta from typing import Optional, Dict, Any @@ -26,8 +25,6 @@ from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard from bot.utils.text_sanitizer import sanitize_display_name, username_for_display from bot.utils.config_link import prepare_config_links -payment_processing_lock = asyncio.Lock() - YOOKASSA_EVENT_PAYMENT_SUCCEEDED = 'payment.succeeded' YOOKASSA_EVENT_PAYMENT_CANCELED = 'payment.canceled' YOOKASSA_EVENT_PAYMENT_WAITING_FOR_CAPTURE = 'payment.waiting_for_capture' @@ -210,6 +207,12 @@ async def process_successful_payment(session: AsyncSession, bot: Bot, try: yk_payment_id_from_hook = payment_info_from_webhook.get("id") + provider_payment_id = str(yk_payment_id_from_hook or "").strip() + if not provider_payment_id: + raise ValueError( + f"Missing provider payment id in successful YooKassa webhook for payment {payment_db_id}" + ) + payment_before_update = None if payment_db_id is not None: payment_before_update = await payment_dal.get_payment_by_db_id( @@ -223,6 +226,19 @@ async def process_successful_payment(session: AsyncSession, bot: Bot, payment_db_id, ) return + + marked = await payment_dal.mark_provider_payment_succeeded_once( + session, + payment_db_id, + provider_payment_id, + ) + if not marked: + logging.info( + "YooKassa webhook: payment %s already processed atomically", + payment_db_id, + ) + return + should_send_lknpd_receipt = bool( lknpd_service and lknpd_service.configured @@ -299,18 +315,6 @@ async def process_successful_payment(session: AsyncSession, bot: Bot, raise Exception( f"Subscription Error: Failed to activate for user {user_id}") - updated_payment_record = await payment_dal.update_payment_status_by_db_id( - session, - payment_db_id=payment_db_id, - new_status=payment_info_from_webhook.get("status", "succeeded"), - yk_payment_id=yk_payment_id_from_hook) - if not updated_payment_record: - logging.error( - f"Failed to update payment record {payment_db_id} for yk_id {yk_payment_id_from_hook}" - ) - raise Exception( - f"DB Error: Could not update payment record {payment_db_id}") - base_subscription_end_date = activation_details['end_date'] final_end_date_for_user = base_subscription_end_date applied_promo_bonus_days = activation_details.get( @@ -619,9 +623,8 @@ async def yookassa_webhook_route(request: web.Request): "payment_method": pm_dict, } - async with payment_processing_lock: - async with async_session_factory() as session: - try: + async with async_session_factory() as session: + try: if notification_object.event == YOOKASSA_EVENT_PAYMENT_SUCCEEDED: if payment_dict_for_processing.get( "paid") and payment_dict_for_processing.get( @@ -726,14 +729,14 @@ async def yookassa_webhook_route(request: web.Request): logging.exception("Failed to cancel bind-only payment auth") except Exception: logging.exception("Failed to handle bind-only waiting_for_capture webhook") - except Exception as e_webhook_db_processing: - await session.rollback() - logging.error( - f"Error processing YooKassa webhook event '{notification_object.event}' " - f"for YK Payment ID {payment_dict_for_processing.get('id')} in DB transaction: {e_webhook_db_processing}", - exc_info=True) - return web.Response( - status=200, text="ok_internal_processing_error_logged") + except Exception as e_webhook_db_processing: + await session.rollback() + logging.error( + f"Error processing YooKassa webhook event '{notification_object.event}' " + f"for YK Payment ID {payment_dict_for_processing.get('id')} in DB transaction: {e_webhook_db_processing}", + exc_info=True) + return web.Response( + status=200, text="ok_internal_processing_error_logged") return web.Response(status=200, text="ok") diff --git a/bot/services/crypto_pay_service.py b/bot/services/crypto_pay_service.py index 6fa5f19..ef6d050 100644 --- a/bot/services/crypto_pay_service.py +++ b/bot/services/crypto_pay_service.py @@ -279,6 +279,11 @@ class CryptoPayService: sale_mode=sale_mode, traffic_gb=traffic_gb if sale_mode == "traffic" else None, ) + if not activation or not activation.get("end_date"): + raise RuntimeError( + f"CryptoPay webhook: activation failed for payment {payment_db_id}" + ) + referral_bonus = None if sale_mode != "traffic": referral_bonus = await referral_service.apply_referral_bonuses_for_payment( diff --git a/bot/services/freekassa_service.py b/bot/services/freekassa_service.py index 09638a8..ef6c427 100644 --- a/bot/services/freekassa_service.py +++ b/bot/services/freekassa_service.py @@ -378,6 +378,10 @@ class FreeKassaService: sale_mode=sale_mode, traffic_gb=months if sale_mode == "traffic" else None, ) + if not activation or not activation.get("end_date"): + raise RuntimeError( + f"FreeKassa webhook: activation failed for payment {payment.payment_id}" + ) referral_bonus = None if sale_mode != "traffic": diff --git a/bot/services/platega_service.py b/bot/services/platega_service.py index 890bd5c..9407c4c 100644 --- a/bot/services/platega_service.py +++ b/bot/services/platega_service.py @@ -272,6 +272,10 @@ class PlategaService: sale_mode=sale_mode, traffic_gb=payment_months if sale_mode == "traffic" else None, ) + if not activation or not activation.get("end_date"): + raise RuntimeError( + f"Platega webhook: activation failed for payment {payment.payment_id}" + ) referral_bonus = None if sale_mode != "traffic": diff --git a/bot/services/stars_service.py b/bot/services/stars_service.py index 0981fec..11bb572 100644 --- a/bot/services/stars_service.py +++ b/bot/services/stars_service.py @@ -144,46 +144,57 @@ class StarsService: payment_record = await payment_dal.get_payment_by_db_id(session, payment_db_id) promo_code_id_from_payment = payment_record.promo_code_id if payment_record else None + activation_details = None + referral_bonus = None try: - await payment_dal.update_provider_payment_and_status( - session, payment_db_id, - message.successful_payment.provider_payment_charge_id, - "succeeded") + provider_payment_id = str( + message.successful_payment.provider_payment_charge_id + or f"stars:{payment_db_id}" + ) + marked = await payment_dal.mark_provider_payment_succeeded_once( + session, + payment_db_id, + provider_payment_id, + ) + if not marked: + logging.info( + "Stars payment %s already processed atomically", + payment_db_id, + ) + return + + activation_details = await self.subscription_service.activate_subscription( + session, + message.from_user.id, + int(months) if sale_mode != "traffic" else 0, + float(stars_amount), + payment_db_id, + promo_code_id_from_payment=promo_code_id_from_payment, + provider="telegram_stars", + sale_mode=sale_mode, + traffic_gb=months if sale_mode == "traffic" else None, + ) + if not activation_details or not activation_details.get("end_date"): + raise RuntimeError( + f"Failed to activate subscription after stars payment {payment_db_id}" + ) + + if sale_mode != "traffic": + referral_bonus = await self.referral_service.apply_referral_bonuses_for_payment( + session, + message.from_user.id, + int(months) or 1, + current_payment_db_id=payment_db_id, + skip_if_active_before_payment=False, + ) await session.commit() except Exception as e_upd: await session.rollback() logging.error( - f"Failed to update stars payment record {payment_db_id}: {e_upd}", + f"Failed to process stars payment record {payment_db_id}: {e_upd}", exc_info=True) return - activation_details = await self.subscription_service.activate_subscription( - session, - message.from_user.id, - int(months) if sale_mode != "traffic" else 0, - float(stars_amount), - payment_db_id, - promo_code_id_from_payment=promo_code_id_from_payment, - provider="telegram_stars", - sale_mode=sale_mode, - traffic_gb=months if sale_mode == "traffic" else None, - ) - if not activation_details or not activation_details.get("end_date"): - logging.error( - f"Failed to activate subscription after stars payment for user {message.from_user.id}") - return - - referral_bonus = None - if sale_mode != "traffic": - referral_bonus = await self.referral_service.apply_referral_bonuses_for_payment( - session, - message.from_user.id, - int(months) or 1, - current_payment_db_id=payment_db_id, - skip_if_active_before_payment=False, - ) - await session.commit() - applied_days = referral_bonus.get("referee_bonus_applied_days") if referral_bonus else None final_end = referral_bonus.get("referee_new_end_date") if referral_bonus else None if not final_end: diff --git a/db/alembic_runner.py b/db/alembic_runner.py index 4797420..39d3958 100644 --- a/db/alembic_runner.py +++ b/db/alembic_runner.py @@ -140,6 +140,25 @@ def _run_legacy_migrator_compatibility(connection: Connection) -> None: connection.execute( text("ALTER TABLE promo_codes ADD COLUMN discount_percentage INTEGER") ) + if "current_activations" not in promo_columns: + connection.execute( + text( + "ALTER TABLE promo_codes ADD COLUMN current_activations INTEGER NOT NULL DEFAULT 0" + ) + ) + else: + connection.execute( + text( + "UPDATE promo_codes SET current_activations = 0 " + "WHERE current_activations IS NULL" + ) + ) + connection.execute( + text("ALTER TABLE promo_codes ALTER COLUMN current_activations SET DEFAULT 0") + ) + connection.execute( + text("ALTER TABLE promo_codes ALTER COLUMN current_activations SET NOT NULL") + ) if "bonus_days" in promo_columns: connection.execute( text("ALTER TABLE promo_codes ALTER COLUMN bonus_days DROP NOT NULL")