add WATA provider / broadcast retry / cluster sync concurrency / gift fix

handlers/admin/clusters/cluster_sync.py

@@ -36,6 +36,124 @@ from .base import router
 from .keyboard import AdminClusterCallback, build_availability_kb, build_sync_cluster_kb
 
 
+SYNC_CONCURRENCY = 200
+
+
+async def _fetch_all_panel_uuids(remna: RemnawaveAPI) -> set[str]:
+    uuids: set[str] = set()
+    page_size = 1000
+    start = 0
+
+    while True:
+        try:
+            r = await remna._request("GET", "/users", params={"size": page_size, "start": start})
+        except Exception as e:
+            logger.error(f"[Sync] GET /users error at start={start}: {e}")
+            break
+
+        if r.status_code != 200:
+            logger.error(f"[Sync] GET /users returned {r.status_code}: {r.text[:200]}")
+            break
+
+        try:
+            raw = r.json()
+        except Exception as e:
+            logger.error(f"[Sync] GET /users JSON parse error: {e}")
+            break
+
+        body = raw.get("response") or raw.get("data") or raw
+        users = body.get("users") or []
+        total = int(body.get("total") or 0)
+
+        if not users:
+            break
+
+        for u in users:
+            uid = u.get("uuid")
+            if uid:
+                uuids.add(str(uid))
+
+        start += len(users)
+        if len(users) < page_size or (total and start >= total):
+            break
+
+    return uuids
+
+
+def _compute_user_fields(
+    key,
+    tariff: dict | None,
+    cluster_servers: list[dict],
+    use_country_selection: bool,
+) -> tuple[int, int, str | None, list[str]]:
+    traffic_limit_bytes = 0
+    hwid_limit = 0
+    subgroup_title = tariff.get("subgroup_title") if tariff else None
+
+    current_device_limit_from_key = key.get("current_device_limit")
+    current_traffic_limit_gb_from_key = key.get("current_traffic_limit")
+    selected_device_limit_from_key = key.get("selected_device_limit")
+    selected_traffic_limit_gb_from_key = key.get("selected_traffic_limit")
+
+    if tariff:
+        if current_traffic_limit_gb_from_key is not None:
+            traffic_limit_bytes = int(current_traffic_limit_gb_from_key * 1024**3)
+        elif selected_traffic_limit_gb_from_key is not None:
+            traffic_limit_bytes = int(selected_traffic_limit_gb_from_key * 1024**3)
+        elif tariff.get("traffic_limit") is not None:
+            traffic_limit_bytes = int(tariff.get("traffic_limit") * 1024**3)
+
+        if current_device_limit_from_key is not None:
+            hwid_limit = int(current_device_limit_from_key)
+        elif selected_device_limit_from_key is not None:
+            hwid_limit = int(selected_device_limit_from_key)
+        else:
+            hwid_limit = int(tariff.get("device_limit") or 0)
+
+    expire_iso: str | None = None
+    if key.get("expiry_time"):
+        expire_iso = (
+            datetime.utcfromtimestamp(key["expiry_time"] / 1000)
+            .replace(tzinfo=timezone.utc)
+            .isoformat()
+        )
+
+    if use_country_selection:
+        user_server = None
+        for s in cluster_servers:
+            if s.get("server_name") == key["server_id"]:
+                user_server = s
+                break
+        inbound_ids = (
+            [user_server["inbound_id"]] if user_server and user_server.get("inbound_id") else []
+        )
+    else:
+        filtered_servers = cluster_servers
+        if subgroup_title or (tariff and tariff.get("id")):
+            tid = tariff.get("id") if tariff else None
+            filtered_servers = [
+                s
+                for s in cluster_servers
+                if (tid and tid in (s.get("tariff_ids") or []))
+                or (subgroup_title and subgroup_title in (s.get("tariff_subgroups") or []))
+            ]
+            if not filtered_servers:
+                filtered_servers = cluster_servers
+
+        if tariff and tariff.get("group_code"):
+            group_code = tariff.get("group_code").lower()
+            if group_code in ALLOWED_GROUP_CODES:
+                special_filtered = [
+                    s for s in filtered_servers if group_code in (s.get("special_groups") or [])
+                ]
+                if special_filtered:
+                    filtered_servers = special_filtered
+
+        inbound_ids = [s["inbound_id"] for s in filtered_servers if s.get("inbound_id")]
+
+    return traffic_limit_bytes, hwid_limit, expire_iso, inbound_ids
+
+
 @router.callback_query(AdminClusterCallback.filter(F.action == "availability"), IsAdminFilter())
 async def handle_cluster_availability(
     callback_query: types.CallbackQuery,
@@ -509,180 +627,227 @@ async def handle_sync_cluster(
             tariffs_cache = {t.id: dict(t.__dict__) for t in tariffs_list}
 
         if only_remnawave:
-            batch_size = 250
             total_keys = len(keys_to_sync)
-            processed_count = 0
 
-            for batch_start in range(0, total_keys, batch_size):
-                batch = keys_to_sync[batch_start : batch_start + batch_size]
-                batch_end = batch_start + len(batch)
-                logger.info(f"[Sync] Обработка батча {batch_start}-{batch_end} из {total_keys}")
+            api_url = cluster_servers[0]["api_url"]
+            remna = RemnawaveAPI(api_url)
+            login_ok = await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD)
+            if not login_ok:
+                await callback_query.message.edit_text(
+                    text=f"❌ Не удалось авторизоваться в Remnawave для кластера {cluster_name}.",
+                    reply_markup=build_admin_back_kb("clusters"),
+                )
+                return
 
-                async def update_remnawave_api(key):
-                    try:
-                        traffic_limit_bytes = 0
-                        hwid_limit = 0
-                        subgroup_title = None
-                        tariff = tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None
+            try:
+                await callback_query.message.edit_text(
+                    text=(
+                        f"<b>🔄 Синхронизация кластера {cluster_name}</b>\n\n"
+                        f"🔑 Ключей в БД: <b>{total_keys}</b>\n\n"
+                        "📥 Получение списка пользователей с панели..."
+                    )
+                )
+                panel_uuids = await _fetch_all_panel_uuids(remna)
+                logger.info(f"[Sync] В панели {cluster_name}: {len(panel_uuids)} юзеров")
 
-                        current_device_limit_from_key = key.get("current_device_limit")
-                        current_traffic_limit_gb_from_key = key.get("current_traffic_limit")
-                        selected_device_limit_from_key = key.get("selected_device_limit")
-                        selected_traffic_limit_gb_from_key = key.get("selected_traffic_limit")
+                to_update: list = []
+                to_create: list = []
+                for key in keys_to_sync:
+                    if str(key["client_id"]) in panel_uuids:
+                        to_update.append(key)
+                    else:
+                        to_create.append(key)
 
-                        if tariff:
-                            if current_traffic_limit_gb_from_key is not None:
-                                traffic_limit_bytes = int(current_traffic_limit_gb_from_key * 1024**3)
-                            elif selected_traffic_limit_gb_from_key is not None:
-                                traffic_limit_bytes = int(selected_traffic_limit_gb_from_key * 1024**3)
-                            elif tariff.get("traffic_limit") is not None:
-                                traffic_limit_bytes = int(tariff.get("traffic_limit") * 1024**3)
-                            else:
-                                traffic_limit_bytes = 0
+                logger.info(
+                    f"[Sync] {cluster_name}: к update={len(to_update)}, к create={len(to_create)}"
+                )
 
-                            if current_device_limit_from_key is not None:
-                                hwid_limit = int(current_device_limit_from_key)
-                            elif selected_device_limit_from_key is not None:
-                                hwid_limit = int(selected_device_limit_from_key)
-                            else:
-                                hwid_limit = tariff.get("device_limit")
+                semaphore = asyncio.Semaphore(SYNC_CONCURRENCY)
+                pending_db_updates: list[dict] = []
+                pending_lock = asyncio.Lock()
+                stats = {"updated": 0, "created": 0, "failed": 0, "done": 0}
 
-                            subgroup_title = tariff.get("subgroup_title")
-
-                        expire_iso = (
-                            datetime.utcfromtimestamp(key["expiry_time"] / 1000)
-                            .replace(tzinfo=timezone.utc)
-                            .isoformat()
-                        )
-
-                        if use_country_selection:
-                            user_server = None
-                            for s in cluster_servers:
-                                if s.get("server_name") == key["server_id"]:
-                                    user_server = s
-                                    break
-
-                            if not user_server:
-                                return {"key": key, "success": False, "error": "Server not found"}
-
-                            remna = RemnawaveAPI(user_server["api_url"])
-                            inbound_ids = [user_server["inbound_id"]] if user_server.get("inbound_id") else []
-                        else:
-                            remna = RemnawaveAPI(cluster_servers[0]["api_url"])
-
-                            filtered_servers = cluster_servers
-                            if subgroup_title or (tariff and tariff.get("id")):
-                                tid = tariff.get("id") if tariff else None
-                                filtered_servers = [
-                                    s
-                                    for s in cluster_servers
-                                    if (tid and tid in (s.get("tariff_ids") or []))
-                                    or (subgroup_title and subgroup_title in (s.get("tariff_subgroups") or []))
-                                ]
-                                if not filtered_servers:
-                                    filtered_servers = cluster_servers
-
-                            if tariff and tariff.get("group_code"):
-                                group_code = tariff.get("group_code").lower()
-                                if group_code in ALLOWED_GROUP_CODES:
-                                    special_filtered = [
-                                        s for s in filtered_servers if group_code in (s.get("special_groups") or [])
-                                    ]
-                                    if special_filtered:
-                                        filtered_servers = special_filtered
-
-                            inbound_ids = [s["inbound_id"] for s in filtered_servers if s.get("inbound_id")]
-
-                        if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-                            return {"key": key, "success": False, "error": "Login failed"}
-
-                        success = await remna.update_user(
-                            uuid=key["client_id"],
-                            expire_at=expire_iso,
-                            telegram_id=int(key.get("owner_tg_id") or 0),
-                            email=f"{key['email']}@fake.local",
-                            active_user_inbounds=inbound_ids,
-                            traffic_limit_bytes=traffic_limit_bytes,
-                            hwid_device_limit=hwid_limit,
-                        )
-
-                        if success:
-                            sub = await remna.get_subscription_by_username(key["email"])
-                            new_link = sub.get("subscriptionUrl") if sub else None
-                            return {
-                                "key": key,
-                                "success": True,
-                                "new_link": new_link,
-                                "tariff": tariff,
-                                "traffic_limit_bytes": traffic_limit_bytes,
-                                "hwid_limit": hwid_limit,
-                            }
-                        else:
-                            return {
-                                "key": key,
-                                "success": False,
-                                "needs_recreate": True,
-                                "tariff": tariff,
-                                "traffic_limit_bytes": traffic_limit_bytes,
-                                "hwid_limit": hwid_limit,
-                            }
-
-                    except Exception as e:
-                        logger.error(f"[Sync] Ошибка API для {key.get('email')}: {e}")
-                        return {"key": key, "success": False, "error": str(e)}
-
-                tasks = [update_remnawave_api(key) for key in batch]
-                results = await asyncio.gather(*tasks, return_exceptions=True)
-
-                bulk_updates = []
-                recreate_tasks = []
-
-                for result in results:
-                    if isinstance(result, Exception):
-                        logger.error(f"[Sync] Exception в батче: {result}")
-                        continue
-
-                    if not isinstance(result, dict):
-                        continue
-
-                    key = result.get("key")
-                    if not key:
-                        continue
-
-                    try:
-                        if result.get("success") and result.get("new_link"):
-                            new_link = result["new_link"]
-                            tariff = result.get("tariff")
-
-                            key_value = await make_aggregated_link(
-                                session=session,
-                                cluster_all=cluster_servers,
-                                cluster_id=cluster_name,
-                                email=key["email"],
-                                client_id=key["client_id"],
-                                tg_id=key["user_id"],
-                                remna_link_override=None,
-                                plan=tariff,
+                async def process_update(key):
+                    async with semaphore:
+                        try:
+                            tariff = (
+                                tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None
+                            )
+                            traffic_limit_bytes, hwid_limit, expire_iso, inbound_ids = (
+                                _compute_user_fields(
+                                    key, tariff, cluster_servers, use_country_selection
+                                )
                             )
 
-                            bulk_updates.append({
-                                "client_id": key["client_id"],
-                                "remnawave_link": new_link,
-                                "key": key_value,
-                            })
+                            if use_country_selection and not inbound_ids:
+                                stats["failed"] += 1
+                                logger.warning(
+                                    f"[Sync] update {key.get('email')}: server not found"
+                                )
+                                return
 
-                        elif result.get("needs_recreate"):
-                            recreate_tasks.append((key, result))
+                            success = await remna.update_user(
+                                uuid=key["client_id"],
+                                expire_at=expire_iso,
+                                telegram_id=int(key.get("owner_tg_id") or 0),
+                                email=f"{key['email']}@fake.local",
+                                active_user_inbounds=inbound_ids,
+                                traffic_limit_bytes=traffic_limit_bytes,
+                                hwid_device_limit=hwid_limit,
+                            )
 
+                            if not success:
+                                stats["failed"] += 1
+                                logger.warning(f"[Sync] update_user failed for {key.get('email')}")
+                                return
+
+                            sub = await remna.get_subscription_by_username(key["email"])
+                            new_link = sub.get("subscriptionUrl") if sub else None
+
+                            stats["updated"] += 1
+                            if new_link:
+                                async with pending_lock:
+                                    pending_db_updates.append({
+                                        "key": key,
+                                        "tariff": tariff,
+                                        "new_link": new_link,
+                                    })
+                        except Exception as e:
+                            stats["failed"] += 1
+                            logger.error(f"[Sync] update error for {key.get('email')}: {e}")
+                        finally:
+                            stats["done"] += 1
+
+                async def process_create(key):
+                    async with semaphore:
+                        try:
+                            tariff = (
+                                tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None
+                            )
+                            traffic_limit_bytes, hwid_limit, expire_iso, inbound_ids = (
+                                _compute_user_fields(
+                                    key, tariff, cluster_servers, use_country_selection
+                                )
+                            )
+
+                            if not expire_iso:
+                                stats["failed"] += 1
+                                logger.warning(
+                                    f"[Sync] create {key.get('email')}: no expiry_time"
+                                )
+                                return
+
+                            payload = {
+                                "uuid": str(key["client_id"]),
+                                "username": key["email"],
+                                "expireAt": expire_iso,
+                                "status": "ACTIVE",
+                                "trafficLimitStrategy": "NO_RESET",
+                                "trafficLimitBytes": traffic_limit_bytes,
+                                "hwidDeviceLimit": hwid_limit,
+                                "email": f"{key['email']}@fake.local",
+                            }
+                            if key.get("owner_tg_id"):
+                                payload["telegramId"] = int(key["owner_tg_id"])
+                            if inbound_ids:
+                                payload["activeInternalSquads"] = inbound_ids
+
+                            r = await remna._request("POST", "/users", json=payload)
+                            if r.status_code not in (200, 201):
+                                stats["failed"] += 1
+                                logger.warning(
+                                    f"[Sync] create failed for {key.get('email')}: "
+                                    f"{r.status_code} {r.text[:200]}"
+                                )
+                                return
+
+                            sub = await remna.get_subscription_by_username(key["email"])
+                            new_link = sub.get("subscriptionUrl") if sub else None
+
+                            stats["created"] += 1
+                            if new_link:
+                                async with pending_lock:
+                                    pending_db_updates.append({
+                                        "key": key,
+                                        "tariff": tariff,
+                                        "new_link": new_link,
+                                    })
+                        except Exception as e:
+                            stats["failed"] += 1
+                            logger.error(f"[Sync] create error for {key.get('email')}: {e}")
+                        finally:
+                            stats["done"] += 1
+
+                async def progress_loop():
+                    while True:
+                        await asyncio.sleep(3)
+                        done = stats["done"]
+                        if total_keys == 0:
+                            return
+                        percent = int((done / total_keys) * 100)
+                        bar = "█" * (percent // 5) + "░" * (20 - percent // 5)
+                        try:
+                            await callback_query.message.edit_text(
+                                text=(
+                                    f"<b>🔄 Синхронизация кластера {cluster_name}</b>\n\n"
+                                    f"🔑 Всего: <b>{total_keys}</b> "
+                                    f"(update: {len(to_update)}, create: {len(to_create)})\n\n"
+                                    f"Готово: <b>{done}/{total_keys}</b> ({percent}%)\n"
+                                    f"<code>{bar}</code>\n\n"
+                                    f"✏️ Обновлено: {stats['updated']}\n"
+                                    f"➕ Создано: {stats['created']}\n"
+                                    f"❌ Ошибок: {stats['failed']}"
+                                )
+                            )
+                        except Exception:
+                            pass
+
+                progress_task = asyncio.create_task(progress_loop())
+                try:
+                    tasks = [process_update(k) for k in to_update] + [
+                        process_create(k) for k in to_create
+                    ]
+                    await asyncio.gather(*tasks, return_exceptions=True)
+                finally:
+                    progress_task.cancel()
+                    try:
+                        await progress_task
+                    except (asyncio.CancelledError, Exception):
+                        pass
+
+                logger.info(
+                    f"[Sync] HTTP-фаза завершена: updated={stats['updated']}, "
+                    f"created={stats['created']}, failed={stats['failed']}"
+                )
+
+                bulk_updates: list[dict] = []
+                for item in pending_db_updates:
+                    key = item["key"]
+                    try:
+                        key_value = await make_aggregated_link(
+                            session=session,
+                            cluster_all=cluster_servers,
+                            cluster_id=cluster_name,
+                            email=key["email"],
+                            client_id=key["client_id"],
+                            tg_id=key["user_id"],
+                            remna_link_override=None,
+                            plan=item["tariff"],
+                        )
+                        bulk_updates.append({
+                            "client_id": key["client_id"],
+                            "remnawave_link": item["new_link"],
+                            "key": key_value,
+                        })
                     except Exception as e:
-                        logger.error(f"[Sync] Ошибка подготовки для {key.get('email')}: {e}")
+                        logger.error(f"[Sync] make_aggregated_link error for {key.get('email')}: {e}")
 
                 if bulk_updates:
                     try:
                         await session.run_sync(
                             lambda sync_session: sync_session.bulk_update_mappings(Key, bulk_updates)
                         )
-                        logger.info(f"[Sync] Bulk: обновлено {len(bulk_updates)} ключей")
+                        logger.info(f"[Sync] Bulk: обновлено {len(bulk_updates)} ключей в БД")
                     except Exception as bulk_error:
                         logger.warning(f"[Sync] Bulk упал, fallback: {bulk_error}")
                         await session.rollback()
@@ -692,56 +857,17 @@ async def handle_sync_cluster(
                                 await session.execute(
                                     update(Key)
                                     .where(Key.client_id == upd["client_id"])
-                                    .values(remnawave_link=upd["remnawave_link"], key=upd["key"])
+                                    .values(
+                                        remnawave_link=upd["remnawave_link"], key=upd["key"]
+                                    )
                                 )
                             except Exception as e:
-                                logger.error(f"[Sync] Fallback ошибка {upd['client_id']}: {e}")
+                                logger.error(
+                                    f"[Sync] Fallback ошибка {upd['client_id']}: {e}"
+                                )
                                 await session.rollback()
-
-                for key, result in recreate_tasks:
-                    try:
-                        logger.warning(f"[Sync] Пересоздание {key['email']}")
-                        await delete_key_from_cluster(cluster_name, key["email"], key["client_id"], session)
-                        await session.execute(
-                            delete(Key).where(Key.user_id == key["user_id"], Key.client_id == key["client_id"])
-                        )
-
-                        cluster_id_for_recreate = key["server_id"] if use_country_selection else cluster_name
-                        await create_key_on_cluster(
-                            cluster_id_for_recreate,
-                            key["user_id"],
-                            key["client_id"],
-                            key["email"],
-                            key["expiry_time"],
-                            plan=key["tariff_id"],
-                            session=session,
-                            remnawave_link=key["remnawave_link"],
-                            hwid_limit=result.get("hwid_limit"),
-                            traffic_limit_bytes=result.get("traffic_limit_bytes"),
-                            selected_device_limit=key.get("selected_device_limit"),
-                            selected_traffic_limit_gb=key.get("selected_traffic_limit"),
-                            current_device_limit=key.get("current_device_limit"),
-                            current_traffic_limit_gb=key.get("current_traffic_limit"),
-                            selected_price_rub=key.get("selected_price_rub"),
-                        )
-                    except Exception as e:
-                        logger.error(f"[Sync] Пересоздание ошибка {key.get('email')}: {e}")
-
-                processed_count = batch_end
-                progress_percent = int((processed_count / total_keys) * 100)
-                progress_bar = "█" * (progress_percent // 5) + "░" * (20 - progress_percent // 5)
-
-                try:
-                    await callback_query.message.edit_text(
-                        text=(
-                            f"<b>🔄 Синхронизация кластера {cluster_name}</b>\n\n"
-                            f"🔑 Количество ключей: <b>{total_keys}</b>\n\n"
-                            f"Обработано: <b>{processed_count}/{total_keys}</b>\n"
-                            f"<code>{progress_bar}</code>"
-                        )
-                    )
-                except Exception:
-                    pass
+            finally:
+                await remna.aclose()
 
         else:
             for key in keys_to_sync:

handlers/admin/sender/scheduled_service.py

@@ -35,7 +35,7 @@ def clamp_broadcast_workers(value: int | None) -> int:
 
 
 def clamp_broadcast_rate(value: int | None) -> int:
-    return max(1, min(int(value or 30), 60))
+    return max(1, min(int(value or 25), 60))
 
 
 def ensure_utc_datetime(value: datetime) -> datetime:

handlers/admin/sender/sender_handler.py

@@ -47,7 +47,7 @@ from .sender_states import AdminSender
 from .sender_utils import get_recipients, parse_message_buttons
 
 
-def _broadcast_progress_text(completed: int, total: int, sent: int, failed: int) -> str:
+def _broadcast_progress_text(completed: int, total: int, sent: int, failed: int, pending: int = 0) -> str:
     """Формирует текст статус-бара рассылки."""
     if total <= 0:
         pct = 0
@@ -56,7 +56,10 @@ def _broadcast_progress_text(completed: int, total: int, sent: int, failed: int)
         pct = min(100, int(100 * completed / total))
         bar_filled = min(10, int(10 * completed / total))
     bar = "█" * bar_filled + "░" * (10 - bar_filled)
-    return f"📤 <b>Рассылка...</b>\n\n[{bar}] <b>{pct}%</b> ({completed}/{total})\n✅ {sent}   ❌ {failed}"
+    base = f"📤 <b>Рассылка...</b>\n\n[{bar}] <b>{pct}%</b> ({completed}/{total})\n✅ {sent}   ❌ {failed}"
+    if pending > 0:
+        base += f"   🔄 {pending}"
+    return base
 
 
 def _compose_message_text() -> str:
@@ -359,8 +362,8 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont
     if should_run_heavy_tasks_separately():
         main_loop = asyncio.get_running_loop()
 
-        async def _edit_progress(completed: int, total: int, sent: int, failed: int) -> None:
-            text = _broadcast_progress_text(completed, total, sent, failed)
+        async def _edit_progress(completed: int, total: int, sent: int, failed: int, pending: int) -> None:
+            text = _broadcast_progress_text(completed, total, sent, failed, pending)
             try:
                 await bot.edit_message_text(
                     chat_id=status_message.chat.id,
@@ -371,10 +374,10 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont
                 if "message is not modified" not in str(e).lower():
                     logger.debug(f"[Sender] Обновление прогресса: {e}")
 
-        def progress_cb(completed: int, total: int, sent: int, failed: int) -> None:
+        def progress_cb(completed: int, total: int, sent: int, failed: int, pending: int) -> None:
             main_loop.call_soon_threadsafe(
-                lambda c=completed, t=total, s=sent, f=failed: asyncio.ensure_future(
-                    _edit_progress(c, t, s, f), loop=main_loop
+                lambda c=completed, t=total, s=sent, f=failed, p=pending: asyncio.ensure_future(
+                    _edit_progress(c, t, s, f, p), loop=main_loop
                 )
             )
 
@@ -398,8 +401,8 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont
             message_data = {"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard}
             messages.append(message_data)
 
-        async def on_progress(completed: int, total: int, sent: int, failed: int) -> None:
-            text = _broadcast_progress_text(completed, total, sent, failed)
+        async def on_progress(completed: int, total: int, sent: int, failed: int, pending: int) -> None:
+            text = _broadcast_progress_text(completed, total, sent, failed, pending)
             try:
                 await bot.edit_message_text(
                     chat_id=status_message.chat.id,
@@ -410,7 +413,7 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont
                 if "message is not modified" not in str(e).lower():
                     logger.debug(f"[Sender] Обновление прогресса: {e}")
 
-        broadcast_service = BroadcastService(bot=bot, session=session, messages_per_second=30)
+        broadcast_service = BroadcastService(bot=bot, session=session)
         stats = await broadcast_service.broadcast(
             messages,
             workers=5,
@@ -464,7 +467,7 @@ async def handle_schedule_datetime_input(message: Message, state: FSMContext, se
         keyboard_json=data.get("keyboard"),
         scheduled_for=scheduled_for,
         workers=5,
-        messages_per_second=30,
+        messages_per_second=25,
     )
     await state.clear()
     await message.answer(

handlers/admin/sender/sender_service.py

@@ -16,16 +16,22 @@ from database import async_session_maker, save_blocked_user_ids
 from logger import logger
 
 
+DEFAULT_MESSAGES_PER_SECOND = 25
+MAX_RETRY_ATTEMPTS = 5
+MAX_RETRY_AFTER_SECONDS = 120.0
+
+
 def run_broadcast_in_thread(
     api_token: str,
     tg_ids: list[int],
     text_message: str,
     photo: str | None,
     keyboard_data: dict | None,
-    progress_cb: Callable[[int, int, int, int], None] | None = None,
+    progress_cb: Callable[[int, int, int, int, int], None] | None = None,
 ) -> dict:
     """
     Синхронная обёртка: запускает рассылку в отдельном event loop в текущем потоке.
+    progress_cb принимает (completed, total, sent, failed, pending_retries).
     """
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
@@ -34,11 +40,11 @@ def run_broadcast_in_thread(
         bot = Bot(token=api_token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
         keyboard = InlineKeyboardMarkup.model_validate(keyboard_data) if keyboard_data else None
         messages = [{"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard} for tg_id in tg_ids]
-        service = BroadcastService(bot=bot, session=None, messages_per_second=30)
+        service = BroadcastService(bot=bot, session=None, messages_per_second=DEFAULT_MESSAGES_PER_SECOND)
 
-        async def on_progress(completed: int, total: int, sent: int, failed: int) -> None:
+        async def on_progress(completed: int, total: int, sent: int, failed: int, pending: int) -> None:
             if progress_cb:
-                progress_cb(completed, total, sent, failed)
+                progress_cb(completed, total, sent, failed, pending)
 
         return loop.run_until_complete(
             service.broadcast(
@@ -58,17 +64,19 @@ def run_broadcast_in_thread(
 
 
 class BroadcastMessage:
+    __slots__ = ("tg_id", "text", "photo", "keyboard", "attempts", "retry_at")
+
     def __init__(self, tg_id: int, text: str, photo: str | None = None, keyboard: Any = None) -> None:
         self.tg_id = tg_id
         self.text = text
         self.photo = photo
         self.keyboard = keyboard
-        self.retry_after = None
         self.attempts = 0
+        self.retry_at = 0.0
 
 
 class RateLimiter:
-    def __init__(self, max_rate: int = 30, window: float = 1.0) -> None:
+    def __init__(self, max_rate: int = DEFAULT_MESSAGES_PER_SECOND, window: float = 1.0) -> None:
         self.max_rate = max_rate
         self.window = window
         self.send_times = deque()
@@ -102,47 +110,57 @@ class BroadcastService:
         self,
         bot: Bot,
         session: AsyncSession | None = None,
-        messages_per_second: int = 30,
+        messages_per_second: int = DEFAULT_MESSAGES_PER_SECOND,
+        max_attempts: int = MAX_RETRY_ATTEMPTS,
     ) -> None:
         self.bot = bot
         self._session = session
         self.rate_limiter = RateLimiter(max_rate=messages_per_second)
-        self.blocked_users = set()
-        self.queue = asyncio.Queue()
-        self.delayed_queue = asyncio.Queue()
-        self.results = []
+        self.max_attempts = max_attempts
+        self.blocked_users: set[int] = set()
+        self.queue: asyncio.Queue = asyncio.Queue()
+        self.results: list[bool] = []
         self.total_sent = 0
-        self.start_time = None
+        self.pending_retries = 0
+        self.start_time: float | None = None
         self.is_running = False
 
-    async def _send_single_message(self, msg: BroadcastMessage) -> bool:
+    async def _send_single_message(self, msg: BroadcastMessage) -> str:
         try:
             await self.rate_limiter.acquire()
 
             if msg.photo:
                 await self.bot.send_photo(
-                    chat_id=msg.tg_id, photo=msg.photo, caption=msg.text, parse_mode="HTML", reply_markup=msg.keyboard
+                    chat_id=msg.tg_id,
+                    photo=msg.photo,
+                    caption=msg.text,
+                    parse_mode="HTML",
+                    reply_markup=msg.keyboard,
                 )
             else:
                 await self.bot.send_message(
-                    chat_id=msg.tg_id, text=msg.text, parse_mode="HTML", reply_markup=msg.keyboard
+                    chat_id=msg.tg_id,
+                    text=msg.text,
+                    parse_mode="HTML",
+                    reply_markup=msg.keyboard,
                 )
 
-            return True
+            return "ok"
 
         except TelegramRetryAfter as e:
-            msg.retry_after = e.retry_after
+            wait_seconds = min(float(e.retry_after), MAX_RETRY_AFTER_SECONDS)
             msg.attempts += 1
+            msg.retry_at = time.time() + wait_seconds
             logger.warning(
-                f"⚠️ Flood control для {msg.tg_id}: повтор через {e.retry_after} сек. (попытка {msg.attempts})"
+                f"⚠️ Flood control для {msg.tg_id}: повтор через {wait_seconds:.0f} сек "
+                f"(попытка {msg.attempts}/{self.max_attempts})"
             )
-            await self.delayed_queue.put(msg)
-            return False
+            return "retry"
 
         except TelegramForbiddenError:
             logger.warning(f"🚫 Бот заблокирован пользователем {msg.tg_id}")
             self.blocked_users.add(msg.tg_id)
-            return False
+            return "fail"
 
         except TelegramBadRequest as e:
             error_msg = str(e).lower()
@@ -151,56 +169,66 @@ class BroadcastService:
                 self.blocked_users.add(msg.tg_id)
             else:
                 logger.warning(f"📩 Не удалось отправить сообщение пользователю {msg.tg_id}: {e}")
-            return False
+            return "fail"
 
         except Exception as e:
             logger.error(f"❌ Ошибка отправки сообщения пользователю {msg.tg_id}: {e}")
-            return False
+            return "fail"
 
-    async def _process_delayed_messages(self):
-        while self.is_running:
+    async def _schedule_retry(self, msg: BroadcastMessage) -> None:
+        try:
+            wait = msg.retry_at - time.time()
+            if wait > 0:
+                await asyncio.sleep(wait)
+            await self.queue.put(msg)
+        except asyncio.CancelledError:
+            self.results.append(False)
+            raise
+        except Exception as e:
+            logger.error(f"❌ Ошибка retry-планировщика для {msg.tg_id}: {e}")
+            self.results.append(False)
+        finally:
+            self.pending_retries -= 1
+
+    async def _worker(self) -> None:
+        while True:
             try:
-                if not self.delayed_queue.empty():
-                    msg = await asyncio.wait_for(self.delayed_queue.get(), timeout=0.1)
-
-                    if msg.retry_after:
-                        await asyncio.sleep(msg.retry_after)
-                        msg.retry_after = None
-
-                    if msg.attempts < 3:
-                        await self.queue.put(msg)
-                    else:
-                        logger.error(f"❌ Достигнут лимит попыток для {msg.tg_id}")
-                        self.results.append(False)
-                else:
-                    await asyncio.sleep(0.1)
-
-            except TimeoutError:
+                msg = await asyncio.wait_for(self.queue.get(), timeout=0.5)
+            except (TimeoutError, asyncio.TimeoutError):
+                if not self.is_running:
+                    return
                 continue
-            except Exception as e:
-                logger.error(f"❌ Ошибка в обработчике отложенных сообщений: {e}")
-                await asyncio.sleep(0.1)
 
-    async def _worker(self):
-        while self.is_running:
             try:
-                msg = await asyncio.wait_for(self.queue.get(), timeout=0.1)
+                result = await self._send_single_message(msg)
 
-                success = await self._send_single_message(msg)
-
-                if success:
+                if result == "ok":
                     self.total_sent += 1
                     self.results.append(True)
-                elif msg.attempts == 0:
+                elif result == "retry":
+                    if msg.attempts < self.max_attempts:
+                        try:
+                            asyncio.create_task(self._schedule_retry(msg))
+                            self.pending_retries += 1
+                        except Exception as e:
+                            logger.error(
+                                f"❌ Не удалось запланировать повтор для {msg.tg_id}: {e}"
+                            )
+                            self.results.append(False)
+                    else:
+                        logger.error(
+                            f"❌ Достигнут лимит попыток для {msg.tg_id} "
+                            f"(после {msg.attempts} попыток)"
+                        )
+                        self.results.append(False)
+                else:
                     self.results.append(False)
 
-                self.queue.task_done()
-
-            except TimeoutError:
-                continue
             except Exception as e:
                 logger.error(f"❌ Ошибка в воркере рассылки: {e}")
-                await asyncio.sleep(0.1)
+                self.results.append(False)
+            finally:
+                self.queue.task_done()
 
     async def _save_blocked_users(self) -> None:
         if not self.blocked_users:
@@ -220,24 +248,30 @@ class BroadcastService:
     async def _progress_loop(
         self,
         total: int,
-        on_progress: Callable[[int, int, int, int], Awaitable[None]],
+        on_progress: Callable[[int, int, int, int, int], Awaitable[None]],
         interval: float,
         progress_every: int,
     ) -> None:
-        """Периодически вызывает on_progress(completed, total, sent, failed)."""
+        """Периодически вызывает on_progress(completed, total, sent, failed, pending_retries)."""
         last_reported = 0
+        last_emit_ts = time.time()
+        force_interval = 30.0
         while self.is_running:
             await asyncio.sleep(interval)
             if not self.is_running:
                 break
             completed = len(self.results)
-            if completed - last_reported < progress_every:
+            now = time.time()
+            new_results = completed - last_reported
+            if new_results < progress_every and (now - last_emit_ts) < force_interval:
                 continue
             sent = self.total_sent
             failed = completed - sent
+            pending = self.pending_retries
             try:
-                await on_progress(completed, total, sent, failed)
+                await on_progress(completed, total, sent, failed, pending)
                 last_reported = completed
+                last_emit_ts = now
             except Exception as e:
                 logger.debug(f"[Broadcast] Ошибка обновления прогресса: {e}")
 
@@ -245,7 +279,7 @@ class BroadcastService:
         self,
         messages: list[dict],
         workers: int = 20,
-        on_progress: Callable[[int, int, int, int], Awaitable[None]] | None = None,
+        on_progress: Callable[[int, int, int, int, int], Awaitable[None]] | None = None,
         progress_interval: float = 2.0,
         progress_every: int = 50,
     ) -> dict:
@@ -254,6 +288,7 @@ class BroadcastService:
         self.results = []
         self.total_sent = 0
         self.blocked_users = set()
+        self.pending_retries = 0
 
         for msg_data in messages:
             msg = BroadcastMessage(
@@ -264,10 +299,13 @@ class BroadcastService:
             )
             await self.queue.put(msg)
 
-        logger.info(f"📤 Начата рассылка на {len(messages)} пользователей с {workers} воркерами")
-
         total = len(messages)
-        progress_task = None
+        logger.info(
+            f"📤 Начата рассылка на {total} пользователей с {workers} воркерами "
+            f"(rate={self.rate_limiter.max_rate}/сек, max_attempts={self.max_attempts})"
+        )
+
+        progress_task: asyncio.Task | None = None
         if on_progress and total > 0:
             progress_task = asyncio.create_task(
                 self._progress_loop(total, on_progress, progress_interval, max(1, progress_every)),
@@ -275,13 +313,11 @@ class BroadcastService:
 
         worker_tasks = [asyncio.create_task(self._worker()) for _ in range(workers)]
 
-        delayed_task = asyncio.create_task(self._process_delayed_messages())
-
-        await self.queue.join()
-
-        await asyncio.sleep(1)
-        while not self.delayed_queue.empty():
-            await asyncio.sleep(1)
+        while True:
+            await self.queue.join()
+            if self.pending_retries <= 0:
+                break
+            await asyncio.sleep(0.5)
 
         self.is_running = False
 
@@ -298,15 +334,14 @@ class BroadcastService:
                     total,
                     self.total_sent,
                     completed - self.total_sent,
+                    0,
                 )
             except Exception as e:
                 logger.debug(f"[Broadcast] Финальное обновление прогресса: {e}")
 
         for task in worker_tasks:
             task.cancel()
-        delayed_task.cancel()
-
-        await asyncio.gather(*worker_tasks, delayed_task, return_exceptions=True)
+        await asyncio.gather(*worker_tasks, return_exceptions=True)
 
         if self._session is not None:
             await self._save_blocked_users()

handlers/admin/settings/settings_config.py

@@ -44,6 +44,8 @@ PAYMENT_PROVIDER_TITLES: Final[dict[str, str]] = {
     "ROBOKASSA": "Robokassa",
     "KASSAI_CARDS": "KassaAI карты",
     "KASSAI_SBP": "KassaAI СБП",
+    "WATA_RU": "WATA карты РФ / СБП",
+    "WATA_INT": "WATA международные",
     "TRIBUTE": "Tribute",
     "HELEKET": "Heleket",
     "CRYPTOBOT": "CryptoBot",

handlers/buttons.py

@@ -80,9 +80,7 @@ FREEKASSA = "💰 FreeKassa: межд. платежи"
 CRYPTOBOT = "💰 CryptoBot: криптовалюта"
 STARS = "⭐ Оплата Звездами"
 ROBOKASSA = "⭐ RoboKassa"
-WATA = "💳 WATA"
-WATA_RU = "🇷🇺 WATA: Карты РФ"
-WATA_SBP = "🏦 WATA: СБП"
+WATA_RU = "🇷🇺 WATA: Карты РФ / СБП"
 WATA_INT = "🌍 WATA: Международные карты"
 KASSAI = "💳 KassaAI"
 KASSAI_CARDS = "💳 KassaAI: Карты РФ"

handlers/notifications/sender.py

@@ -172,18 +172,23 @@ async def _send_text(
         return False
 
 
+_NOTIFY_MAX_ATTEMPTS = 5
+_NOTIFY_MAX_RETRY_AFTER = 120.0
+
+
 class FastNotificationSender:
-    def __init__(self, bot: Bot, messages_per_second: int = 35) -> None:
+    def __init__(self, bot: Bot, messages_per_second: int = 35, max_attempts: int = _NOTIFY_MAX_ATTEMPTS) -> None:
         self.bot = bot
         self.rate_limiter = NotificationRateLimiter(max_rate=messages_per_second)
+        self.max_attempts = max_attempts
         self.blocked_users: set[int] = set()
         self.queue: asyncio.Queue = asyncio.Queue()
-        self.delayed_queue: asyncio.Queue = asyncio.Queue()
         self.results: list[bool] = []
         self.total_sent = 0
+        self.pending_retries = 0
         self.is_running = False
 
-    async def _send_one(self, msg: dict) -> bool:
+    async def _send_one(self, msg: dict) -> str:
         tg_id = msg["tg_id"]
         try:
             await self.rate_limiter.acquire()
@@ -230,57 +235,70 @@ class FastNotificationSender:
                     chat_id=tg_id, text=processed_text, reply_markup=msg.get("keyboard"),
                     **emoji_kwargs, **text_kwargs,
                 )
-            return True
+            return "ok"
 
         except TelegramRetryAfter as e:
-            msg["_retry_after"] = e.retry_after
+            wait_seconds = min(float(e.retry_after), _NOTIFY_MAX_RETRY_AFTER)
             msg["_attempts"] = msg.get("_attempts", 0) + 1
-            await self.delayed_queue.put(msg)
-            return False
+            msg["_retry_at"] = time.time() + wait_seconds
+            return "retry"
         except TelegramForbiddenError:
             self.blocked_users.add(tg_id)
-            return False
+            return "fail"
         except TelegramBadRequest as e:
             if "chat not found" in str(e).lower():
                 self.blocked_users.add(tg_id)
-            return False
+            return "fail"
         except Exception:
-            return False
+            return "fail"
 
-    async def _process_delayed(self):
-        while self.is_running:
+    async def _schedule_retry(self, msg: dict) -> None:
+        try:
+            wait = msg.get("_retry_at", 0.0) - time.time()
+            if wait > 0:
+                await asyncio.sleep(wait)
+            await self.queue.put(msg)
+        except asyncio.CancelledError:
+            self.results.append(False)
+            raise
+        except Exception:
+            self.results.append(False)
+        finally:
+            self.pending_retries -= 1
+
+    async def _worker(self):
+        while True:
             try:
-                if not self.delayed_queue.empty():
-                    msg = await asyncio.wait_for(self.delayed_queue.get(), timeout=0.1)
-                    if msg.get("_retry_after"):
-                        await asyncio.sleep(msg["_retry_after"])
-                        msg["_retry_after"] = None
-                    if msg.get("_attempts", 0) < 3:
-                        await self.queue.put(msg)
+                msg = await asyncio.wait_for(self.queue.get(), timeout=0.5)
+            except (TimeoutError, asyncio.TimeoutError):
+                if not self.is_running:
+                    return
+                continue
+
+            try:
+                result = await self._send_one(msg)
+                if result == "ok":
+                    self.total_sent += 1
+                    self.results.append(True)
+                elif result == "retry":
+                    if msg.get("_attempts", 0) < self.max_attempts:
+                        try:
+                            asyncio.create_task(self._schedule_retry(msg))
+                            self.pending_retries += 1
+                        except Exception as e:
+                            logger.error(
+                                f"Не удалось запланировать повтор для {msg.get('tg_id')}: {e}"
+                            )
+                            self.results.append(False)
                     else:
                         self.results.append(False)
                 else:
-                    await asyncio.sleep(0.1)
-            except TimeoutError:
-                continue
-            except Exception:
-                await asyncio.sleep(0.1)
-
-    async def _worker(self):
-        while self.is_running:
-            try:
-                msg = await asyncio.wait_for(self.queue.get(), timeout=0.1)
-                success = await self._send_one(msg)
-                if success:
-                    self.total_sent += 1
-                    self.results.append(True)
-                elif msg.get("_attempts", 0) == 0:
                     self.results.append(False)
+            except Exception as e:
+                logger.error(f"Ошибка в воркере уведомлений: {e}")
+                self.results.append(False)
+            finally:
                 self.queue.task_done()
-            except TimeoutError:
-                continue
-            except Exception:
-                await asyncio.sleep(0.1)
 
     async def _save_blocked_users(self):
         if not self.blocked_users:
@@ -301,24 +319,24 @@ class FastNotificationSender:
         self.results = []
         self.total_sent = 0
         self.blocked_users = set()
+        self.pending_retries = 0
         start = time.time()
 
         for msg in messages:
             await self.queue.put(msg)
 
         worker_tasks = [asyncio.create_task(self._worker()) for _ in range(workers)]
-        delayed_task = asyncio.create_task(self._process_delayed())
 
-        await self.queue.join()
-        await asyncio.sleep(0.5)
-        while not self.delayed_queue.empty():
+        while True:
+            await self.queue.join()
+            if self.pending_retries <= 0:
+                break
             await asyncio.sleep(0.5)
 
         self.is_running = False
         for task in worker_tasks:
             task.cancel()
-        delayed_task.cancel()
-        await asyncio.gather(*worker_tasks, delayed_task, return_exceptions=True)
+        await asyncio.gather(*worker_tasks, return_exceptions=True)
         await self._save_blocked_users()
 
         duration = time.time() - start

handlers/payments/__init__.py

@@ -27,6 +27,7 @@ from .pay import router as pay_router
 from .robokassa import router as robokassa_router
 from .stars import router as stars_router
 from .tribute import router as tribute_router
+from .wata import router as wata_router
 from .yookassa import router as yookassa_router
 from .yoomoney import router as yoomoney_router
 
@@ -51,6 +52,8 @@ if PROVIDERS.get("KASSAI_CARDS", {}).get("enabled") or PROVIDERS.get("KASSAI_SBP
     router.include_router(kassai_router)
 if PROVIDERS.get("HELEKET", {}).get("enabled"):
     router.include_router(heleket_router)
+if PROVIDERS.get("WATA_RU", {}).get("enabled") or PROVIDERS.get("WATA_INT", {}).get("enabled"):
+    router.include_router(wata_router)
 
 router.include_router(tribute_router)
 router.include_router(gift_router)

handlers/payments/currency_flow.py

@@ -61,10 +61,10 @@ def filter_providers_by_currency(
     for p in providers:
         up = p.upper()
         if currency == "RUB":
-            if up in rub_set or up == "WATA":
+            if up in rub_set:
                 out.append(p)
         elif currency == "USD":
-            if (up not in rub_set or up == "WATA") and up != "STARS":
+            if up not in rub_set and up != "STARS":
                 out.append(p)
         elif currency == "STARS":
             if up == "STARS":
@@ -79,8 +79,6 @@ def currency_for_provider(up_provider: str, rub_providers: Iterable[str]) -> str
         return "RUB"
     if up_provider == "STARS":
         return "STARS"
-    if up_provider == "WATA":
-        return None
     return "USD"
 
 

handlers/payments/wata/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ("router",)
+
+from .handlers import router

handlers/payments/wata/handlers.py

@@ -0,0 +1,144 @@
+from aiogram import F, Router, types
+from aiogram.fsm.context import FSMContext
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from database import get_temporary_data
+from database.models import User
+from handlers.buttons import MAIN_MENU, PAY_2
+from handlers.texts import DEFAULT_PAYMENT_MESSAGE
+from handlers.utils import edit_or_send_message
+from logger import logger
+from services.payments.currency_rates import format_for_user
+
+from ..constants import ALLOWED_TEMP_PAYMENT_STATES
+from .service import (
+    WATA_METHODS,
+    WATA_MIN_AMOUNTS,
+    generate_wata_payment_link,
+    process_callback_pay_wata,
+    router as service_router,
+)
+
+
+router = Router(name="wata_router")
+router.include_router(service_router)
+
+
+@router.callback_query(F.data == "pay_wata_ru")
+async def handle_pay_wata_ru(
+    callback_query: types.CallbackQuery,
+    state: FSMContext,
+    session: AsyncSession,
+):
+    await process_callback_pay_wata(callback_query, state, session, method_name="ru")
+
+
+@router.callback_query(F.data == "pay_wata_int")
+async def handle_pay_wata_int(
+    callback_query: types.CallbackQuery,
+    state: FSMContext,
+    session: AsyncSession,
+):
+    await process_callback_pay_wata(callback_query, state, session, method_name="int")
+
+
+async def _handle_custom_amount_input_wata(
+    event,
+    session: AsyncSession,
+    method_name: str,
+    pay_button_text: str = PAY_2,
+    main_menu_text: str = MAIN_MENU,
+):
+    message = event.message
+    from_user = event.from_user
+    tg_id = from_user.id
+
+    temp_data = await get_temporary_data(session, tg_id)
+    if not temp_data or temp_data["state"] not in ALLOWED_TEMP_PAYMENT_STATES:
+        await edit_or_send_message(
+            target_message=message,
+            text="❌ Не удалось получить данные для оплаты.",
+        )
+        return
+
+    amount = int(temp_data["data"].get("required_amount", 0))
+    if amount <= 0:
+        await edit_or_send_message(
+            target_message=message,
+            text="❌ Не удалось определить сумму оплаты.",
+        )
+        return
+
+    method = WATA_METHODS.get(method_name)
+    if not method or not method["enable"]:
+        await edit_or_send_message(
+            target_message=message,
+            text="❌ Этот способ оплаты Wata временно недоступен.",
+        )
+        return
+
+    min_amount = WATA_MIN_AMOUNTS.get(method_name, 10)
+    if amount < min_amount:
+        symbol = "$" if method["currency"] == "USD" else "₽"
+        await edit_or_send_message(
+            target_message=message,
+            text=f"❌ Минимальная сумма для оплаты через WATA — {symbol}{min_amount}.",
+        )
+        return
+
+    try:
+        payment_url = await generate_wata_payment_link(amount, tg_id, method, session)
+
+        if not payment_url:
+            await edit_or_send_message(
+                target_message=message,
+                text="❌ Произошла ошибка при создании платежа. Попробуйте позже или выберите другой способ оплаты.",
+            )
+            return
+
+        markup = InlineKeyboardMarkup(
+            inline_keyboard=[
+                [InlineKeyboardButton(text=pay_button_text, url=payment_url)],
+                [InlineKeyboardButton(text=main_menu_text, callback_data="profile")],
+            ]
+        )
+
+        result = await session.execute(select(User.language_code).where(User.tg_id == tg_id))
+        language_code = result.scalar_one_or_none()
+        amount_text = await format_for_user(
+            session,
+            tg_id,
+            float(amount),
+            language_code,
+            force_currency=method["currency"],
+        )
+        text_out = DEFAULT_PAYMENT_MESSAGE.format(amount=amount_text)
+
+        await edit_or_send_message(target_message=message, text=text_out, reply_markup=markup)
+    except Exception as e:
+        logger.error(f"[WATA] Ошибка при создании платежа ({method_name}) для пользователя {tg_id}: {e}")
+        await edit_or_send_message(
+            target_message=message,
+            text="Произошла ошибка при создании платежа. Попробуйте позже.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+
+
+async def handle_custom_amount_input_wata_ru(
+    event,
+    session: AsyncSession,
+    pay_button_text: str = PAY_2,
+    main_menu_text: str = MAIN_MENU,
+):
+    await _handle_custom_amount_input_wata(event, session, "ru", pay_button_text, main_menu_text)
+
+
+async def handle_custom_amount_input_wata_int(
+    event,
+    session: AsyncSession,
+    pay_button_text: str = PAY_2,
+    main_menu_text: str = MAIN_MENU,
+):
+    await _handle_custom_amount_input_wata(event, session, "int", pay_button_text, main_menu_text)

handlers/payments/wata/service.py

@@ -0,0 +1,498 @@
+import time
+
+from decimal import ROUND_HALF_UP, Decimal
+
+import aiohttp
+
+from aiogram import F, Router, types
+from aiogram.fsm.context import FSMContext
+from aiogram.fsm.state import State, StatesGroup
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
+from aiogram.utils.keyboard import InlineKeyboardBuilder
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config import (
+    PROVIDERS_ENABLED,
+    WATA_FAIL_URL,
+    WATA_INT_TOKEN,
+    WATA_RU_TOKEN,
+    WATA_SUCCESS_URL,
+)
+from database import register_pending_payment
+from database.models import User
+from handlers.buttons import BACK, PAY_2, WATA_INT, WATA_RU
+from handlers.payments.keyboards import (
+    build_amounts_keyboard,
+    parse_amount_from_callback,
+    pay_keyboard,
+    payment_options_for_user,
+)
+from handlers.texts import (
+    WATA_INT_DESCRIPTION,
+    WATA_PAYMENT_MESSAGE,
+    WATA_PAYMENT_TITLE,
+    WATA_RU_DESCRIPTION,
+)
+from handlers.utils import edit_or_send_message
+from logger import logger
+from services.payments.currency_rates import format_for_user, get_rub_rate, pick_currency, to_rub
+from services.payments.payment_links import register_payment_creator
+
+
+router = Router()
+
+
+WATA_API_LINKS_URL = "https://api.wata.pro/api/h2h/links"
+
+WATA_MIN_AMOUNTS = {
+    "ru": 10,
+    "int": 1,
+}
+
+
+class ReplenishBalanceWata(StatesGroup):
+    choosing_method = State()
+    choosing_amount = State()
+    waiting_for_payment_confirmation = State()
+    entering_custom_amount = State()
+
+
+WATA_METHODS = {
+    "ru": {
+        "enable": PROVIDERS_ENABLED.get("WATA_RU", False),
+        "currency": "RUB",
+        "token": WATA_RU_TOKEN,
+        "button": WATA_RU,
+        "desc": WATA_RU_DESCRIPTION,
+    },
+    "int": {
+        "enable": PROVIDERS_ENABLED.get("WATA_INT", False),
+        "currency": "USD",
+        "token": WATA_INT_TOKEN,
+        "button": WATA_INT,
+        "desc": WATA_INT_DESCRIPTION,
+    },
+}
+
+
+async def get_user_language(session: AsyncSession, tg_id: int) -> str | None:
+    result = await session.execute(select(User.language_code).where(User.tg_id == tg_id))
+    return result.scalar_one_or_none()
+
+
+@router.callback_query(F.data == "pay_wata")
+async def process_callback_pay_wata(
+    callback_query: types.CallbackQuery,
+    state: FSMContext,
+    session: AsyncSession,
+    method_name: str = None,
+):
+    try:
+        tg_id = callback_query.from_user.id
+        logger.info(f"User {tg_id} initiated Wata payment.")
+        await state.clear()
+
+        if method_name:
+            method = WATA_METHODS.get(method_name)
+            if not method or not method["enable"]:
+                await edit_or_send_message(
+                    target_message=callback_query.message,
+                    text="Ошибка: выбранный способ оплаты недоступен.",
+                    reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+                )
+                return
+
+            language_code = await get_user_language(session, tg_id)
+            opts = await payment_options_for_user(
+                session,
+                tg_id,
+                language_code,
+                force_currency=method["currency"],
+            )
+            builder = build_amounts_keyboard(
+                prefix=f"wata_{method_name}",
+                pattern="{prefix}_amount|{price}",
+                back_cb="balance",
+                custom_cb=f"wata_custom_amount|{method_name}",
+                opts=opts,
+            )
+
+            await edit_or_send_message(
+                target_message=callback_query.message,
+                text=method["desc"],
+                reply_markup=builder,
+            )
+            await state.update_data(
+                wata_method=method_name,
+                message_id=callback_query.message.message_id,
+                chat_id=callback_query.message.chat.id,
+            )
+            await state.set_state(ReplenishBalanceWata.choosing_amount)
+            return
+
+        builder = InlineKeyboardBuilder()
+        for name, method in WATA_METHODS.items():
+            if method["enable"]:
+                builder.row(
+                    InlineKeyboardButton(
+                        text=method["button"],
+                        callback_data=f"wata_method|{name}",
+                    )
+                )
+        builder.row(InlineKeyboardButton(text=BACK, callback_data="balance"))
+
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text="Выберите способ оплаты через WATA:",
+            reply_markup=builder.as_markup(),
+        )
+        await state.update_data(
+            message_id=callback_query.message.message_id,
+            chat_id=callback_query.message.chat.id,
+        )
+        await state.set_state(ReplenishBalanceWata.choosing_method)
+
+    except Exception as e:
+        logger.error(f"Error in process_callback_pay_wata for user {callback_query.message.chat.id}: {e}")
+        await callback_query.answer(
+            "Произошла ошибка при инициализации платежа. Попробуйте позже.",
+            show_alert=True,
+        )
+
+
+@router.callback_query(F.data.startswith("wata_method|"))
+async def process_method_selection(callback_query: types.CallbackQuery, state: FSMContext, session: AsyncSession):
+    method_name = callback_query.data.split("|")[1]
+    method = WATA_METHODS.get(method_name)
+
+    if not method or not method["enable"]:
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text="Ошибка: выбранный способ оплаты недоступен.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    await state.update_data(wata_method=method_name)
+    tg_id = callback_query.from_user.id
+
+    language_code = await get_user_language(session, tg_id)
+    opts = await payment_options_for_user(
+        session,
+        tg_id,
+        language_code,
+        force_currency=method["currency"],
+    )
+    builder = build_amounts_keyboard(
+        prefix=f"wata_{method_name}",
+        pattern="{prefix}_amount|{price}",
+        back_cb="pay_wata",
+        custom_cb=f"wata_custom_amount|{method_name}",
+        opts=opts,
+    )
+
+    await edit_or_send_message(
+        target_message=callback_query.message,
+        text=method["desc"],
+        reply_markup=builder,
+    )
+    await state.update_data(message_id=callback_query.message.message_id, chat_id=callback_query.message.chat.id)
+    await state.set_state(ReplenishBalanceWata.choosing_amount)
+
+
+@router.callback_query(F.data.startswith("wata_custom_amount|"))
+async def process_custom_amount_button(callback_query: types.CallbackQuery, state: FSMContext, session: AsyncSession):
+    method_name = callback_query.data.split("|")[1]
+    await state.update_data(wata_method=method_name)
+
+    builder = InlineKeyboardBuilder()
+    builder.row(InlineKeyboardButton(text=BACK, callback_data=f"pay_wata_{method_name}"))
+
+    language_code = await get_user_language(session, callback_query.from_user.id)
+    currency = pick_currency(language_code)
+
+    currency_text = "рублях (₽)" if currency == "RUB" else "долларах ($)"
+    await edit_or_send_message(
+        target_message=callback_query.message,
+        text=f"Пожалуйста, введите сумму пополнения в {currency_text}.",
+        reply_markup=builder.as_markup(),
+    )
+    await state.set_state(ReplenishBalanceWata.entering_custom_amount)
+
+
+@router.message(ReplenishBalanceWata.entering_custom_amount)
+async def handle_custom_amount_input(message: types.Message, state: FSMContext, session: AsyncSession):
+    data = await state.get_data()
+    method_name = data.get("wata_method")
+    method = WATA_METHODS.get(method_name)
+
+    if not method or not method["enable"]:
+        await edit_or_send_message(
+            target_message=message,
+            text="Ошибка: выбранный способ оплаты недоступен.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    language_code = await get_user_language(session, message.from_user.id)
+    currency = pick_currency(language_code)
+
+    try:
+        user_amount = int(message.text.strip())
+        if user_amount <= 0:
+            raise ValueError
+
+        min_amount = WATA_MIN_AMOUNTS.get(method_name, 10)
+        if currency == "USD":
+            min_amount = 1
+        currency_symbol = "$" if currency == "USD" else "₽"
+
+        if user_amount < min_amount:
+            await edit_or_send_message(
+                target_message=message,
+                text=f"❌ Минимальная сумма для оплаты через WATA — {currency_symbol}{min_amount}.",
+                reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+            )
+            return
+    except Exception:
+        await edit_or_send_message(
+            target_message=message,
+            text="❌ Некорректная сумма. Введите целое число больше 0.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    if currency == "RUB":
+        amount_rub = user_amount
+    else:
+        timeout = aiohttp.ClientTimeout(total=30, connect=10)
+        async with aiohttp.ClientSession(timeout=timeout) as session_http:
+            amount_rub = int(await to_rub(user_amount, "USD", session=session_http))
+
+    await state.update_data(amount=amount_rub)
+    payment_url = await generate_wata_payment_link(amount_rub, message.chat.id, method, session)
+
+    if not payment_url:
+        await edit_or_send_message(
+            target_message=message,
+            text="❌ Произошла ошибка при создании платежа. Попробуйте позже или выберите другой способ оплаты.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    confirm_keyboard = pay_keyboard(payment_url, pay_text=PAY_2, back_cb="balance")
+
+    tg_id = message.from_user.id
+    amount_text = await format_for_user(
+        session, tg_id, float(amount_rub), language_code, force_currency=method["currency"]
+    )
+
+    await edit_or_send_message(
+        target_message=message,
+        text=WATA_PAYMENT_MESSAGE.format(amount=amount_text),
+        reply_markup=confirm_keyboard,
+    )
+
+    await state.set_state(ReplenishBalanceWata.waiting_for_payment_confirmation)
+
+
+@router.callback_query(F.data.startswith("wata_ru_amount|") | F.data.startswith("wata_int_amount|"))
+async def process_amount_selection(callback_query: types.CallbackQuery, state: FSMContext, session: AsyncSession):
+    amount = parse_amount_from_callback(callback_query.data, prefixes=["wata_ru", "wata_int"])
+    if amount is None:
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text="Некорректная сумма.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    method_name = "ru" if callback_query.data.startswith("wata_ru") else "int"
+    method = WATA_METHODS.get(method_name)
+
+    if not method or not method["enable"]:
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text="Ошибка: выбранный способ оплаты недоступен.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    min_amount = WATA_MIN_AMOUNTS.get(method_name, 10)
+    if amount < min_amount:
+        currency_symbol = "$" if method["currency"] == "USD" else "₽"
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text=f"❌ Минимальная сумма для оплаты через WATA — {currency_symbol}{min_amount}.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    await state.update_data(amount=amount)
+    payment_url = await generate_wata_payment_link(amount, callback_query.message.chat.id, method, session)
+
+    if not payment_url:
+        await edit_or_send_message(
+            target_message=callback_query.message,
+            text="❌ Произошла ошибка при создании платежа. Попробуйте позже или выберите другой способ оплаты.",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=[]),
+        )
+        return
+
+    confirm_keyboard = pay_keyboard(payment_url, pay_text=PAY_2, back_cb="balance")
+
+    tg_id = callback_query.from_user.id
+    language_code = await get_user_language(session, tg_id)
+    amount_text = await format_for_user(
+        session, tg_id, float(amount), language_code, force_currency=method["currency"]
+    )
+
+    await edit_or_send_message(
+        target_message=callback_query.message,
+        text=WATA_PAYMENT_MESSAGE.format(amount=amount_text),
+        reply_markup=confirm_keyboard,
+    )
+
+    await state.set_state(ReplenishBalanceWata.waiting_for_payment_confirmation)
+
+
+async def generate_wata_payment_link(
+    amount: int,
+    tg_id: int,
+    method: dict,
+    session: AsyncSession | None = None,
+    *,
+    payment_id: str | None = None,
+    success_url: str | None = None,
+    failure_url: str | None = None,
+    metadata: dict | None = None,
+) -> str | None:
+    token = method.get("token") or ""
+    if not token:
+        logger.error(f"[WATA] Не задан токен для кассы {method.get('currency')}")
+        return None
+
+    currency = str(method.get("currency") or "RUB").upper()
+    unique_order_id = payment_id or f"{int(time.time())}_{tg_id}_{int(amount)}"
+
+    pending_metadata = dict(metadata or {})
+    pending_metadata.setdefault("cassa", "ru" if currency == "RUB" else "int")
+
+    pending_original_amount: float | None = None
+
+    if currency == "RUB":
+        api_amount = float(int(amount))
+    else:
+        try:
+            timeout = aiohttp.ClientTimeout(total=15, connect=10)
+            async with aiohttp.ClientSession(timeout=timeout) as http_rates:
+                rate = await get_rub_rate(currency, session=http_rates)
+            usd_amount = (Decimal(str(amount)) * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+            api_amount = float(usd_amount)
+            pending_original_amount = float(usd_amount)
+            pending_metadata["wata_currency"] = currency
+        except Exception as e:
+            logger.error(f"[WATA] Не удалось сконвертировать {amount} RUB → {currency}: {e}")
+            return None
+
+    body = {
+        "amount": api_amount,
+        "currency": currency,
+        "orderId": unique_order_id,
+        "orderDescription": WATA_PAYMENT_TITLE,
+        "successUrl": success_url or WATA_SUCCESS_URL or "",
+        "failUrl": failure_url or WATA_FAIL_URL or "",
+    }
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+
+    timeout = aiohttp.ClientTimeout(total=60, connect=10)
+    try:
+        async with aiohttp.ClientSession(timeout=timeout) as http_session:
+            async with http_session.post(WATA_API_LINKS_URL, headers=headers, json=body) as resp:
+                if resp.status != 200:
+                    try:
+                        error_json = await resp.json()
+                        logger.error(f"[WATA] API error: status={resp.status}, response={error_json}")
+                    except Exception:
+                        text = await resp.text()
+                        logger.error(f"[WATA] API error: status={resp.status}, non-JSON: {text[:300]}")
+                    return None
+
+                try:
+                    resp_json = await resp.json()
+                except Exception as e:
+                    text = await resp.text()
+                    logger.error(f"[WATA] Не удалось распарсить JSON: {e}, ответ={text[:300]}")
+                    return None
+
+                payment_url = resp_json.get("url")
+                if not payment_url:
+                    logger.error(f"[WATA] В ответе нет поля url: {resp_json}")
+                    return None
+
+                await register_pending_payment(
+                    payment_id=unique_order_id,
+                    tg_id=tg_id,
+                    amount=float(int(amount)),
+                    payment_system="wata",
+                    currency="RUB",
+                    metadata=pending_metadata,
+                    original_amount=pending_original_amount,
+                )
+                logger.info(
+                    f"[WATA] Ссылка создана: tg_id={tg_id}, order_id={unique_order_id}, "
+                    f"rub_amount={amount}, api_amount={api_amount} {currency}"
+                )
+                return payment_url
+    except Exception as e:
+        logger.error(f"[WATA] Ошибка создания платежа: {e}")
+        return None
+
+
+def create_link_factory(method_name: str):
+    async def create_link(
+        session: AsyncSession,
+        tg_id: int,
+        amount: float,
+        currency: str,
+        success_url: str | None,
+        failure_url: str | None,
+        metadata: dict | None,
+    ) -> tuple[str, str | None]:
+        method = WATA_METHODS.get(method_name)
+        if not method or not method.get("enable"):
+            raise ValueError("Способ оплаты Wata недоступен")
+
+        amount_int = int(amount)
+        if amount_int <= 0:
+            raise ValueError("Сумма должна быть больше нуля")
+
+        min_amount = WATA_MIN_AMOUNTS.get(method_name, 10)
+        if amount_int < min_amount:
+            symbol = "$" if method["currency"] == "USD" else "₽"
+            raise ValueError(f"Минимальная сумма Wata — {symbol}{min_amount}")
+
+        payment_id = f"{int(time.time())}_{tg_id}_{amount_int}"
+        url = await generate_wata_payment_link(
+            amount_int,
+            tg_id,
+            method,
+            session,
+            payment_id=payment_id,
+            success_url=success_url,
+            failure_url=failure_url,
+            metadata=metadata,
+        )
+        if not url:
+            raise ValueError("Не удалось создать платёж Wata")
+        return (url, payment_id)
+
+    return create_link
+
+
+register_payment_creator("WATA_RU", create_link_factory("ru"))
+register_payment_creator("WATA_INT", create_link_factory("int"))

services/payments/providers.py

@@ -36,6 +36,13 @@ PROVIDERS_BASE: dict[str, dict[str, Any]] = {
         "module": "kassai",
         "order": 50,
     },
+    "WATA_RU": {
+        "currency": "RUB",
+        "value": "pay_wata_ru",
+        "fast": "handle_custom_amount_input_wata_ru",
+        "module": "wata",
+        "order": 55,
+    },
     "TRIBUTE": {
         "currency": "RUB+USD",
         "value": "pay_tribute",
@@ -48,6 +55,13 @@ PROVIDERS_BASE: dict[str, dict[str, Any]] = {
         "fast": "handle_custom_amount_input_heleket",
         "order": 70,
     },
+    "WATA_INT": {
+        "currency": "USD",
+        "value": "pay_wata_int",
+        "fast": "handle_custom_amount_input_wata_int",
+        "module": "wata",
+        "order": 75,
+    },
     "CRYPTOBOT": {
         "currency": "USD",
         "value": "pay_cryptobot",
@@ -74,6 +88,8 @@ WEB_LINK_PROVIDER_IDS = (
     "ROBOKASSA",
     "KASSAI_CARDS",
     "KASSAI_SBP",
+    "WATA_RU",
+    "WATA_INT",
     "HELEKET",
     "FREEKASSA",
     "CRYPTOBOT",

services/payments/wata/__init__.py

@@ -0,0 +1 @@
+

services/payments/wata/webhook.py

@@ -0,0 +1,230 @@
+import base64
+import json
+
+import aiohttp
+
+from aiohttp import web
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+
+from core.redis_cache import cache_get, cache_set
+from core.webhook_abuse import (
+    get_webhook_client_ip,
+    is_webhook_ip_blocked,
+    record_webhook_signature_failure,
+)
+from database import async_session_maker, get_payment_by_payment_id
+from logger import logger
+from services.payments.pipeline import (
+    ParsedPayment,
+    process_cancelled_payment,
+    process_success_payment,
+)
+
+
+_PROVIDER = "wata"
+
+_WATA_PUBLIC_KEY_URL = "https://api.wata.pro/api/h2h/public-key"
+_WATA_PUBLIC_KEY_CACHE_KEY = "wata:public_key:pem"
+_WATA_PUBLIC_KEY_CACHE_TTL = 6 * 60 * 60
+
+
+async def _get_wata_public_key() -> bytes:
+    try:
+        cached = await cache_get(_WATA_PUBLIC_KEY_CACHE_KEY)
+        if isinstance(cached, str) and cached:
+            return cached.encode()
+    except Exception:
+        pass
+
+    timeout = aiohttp.ClientTimeout(total=10)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        async with session.get(_WATA_PUBLIC_KEY_URL) as resp:
+            data = await resp.json()
+            pem_str = str(data["value"])
+
+    try:
+        await cache_set(_WATA_PUBLIC_KEY_CACHE_KEY, pem_str, _WATA_PUBLIC_KEY_CACHE_TTL)
+    except Exception:
+        pass
+
+    return pem_str.encode()
+
+
+async def _verify_signature(raw_json: bytes, signature: str, public_key_pem: bytes) -> bool:
+    try:
+        public_key = serialization.load_pem_public_key(public_key_pem, backend=default_backend())
+        signature_bytes = base64.b64decode(signature)
+        public_key.verify(signature_bytes, raw_json, padding.PKCS1v15(), hashes.SHA512())
+        return True
+    except Exception as e:
+        logger.error(f"[WATA] Ошибка проверки подписи: {e}")
+        return False
+
+
+def _parse_wata_order_id(order_id: str) -> tuple[int | None, float | None]:
+    if not order_id:
+        return None, None
+    parts = order_id.split("_")
+    if len(parts) >= 3:
+        try:
+            return int(parts[1]), float(parts[2])
+        except (ValueError, TypeError):
+            return None, None
+    if len(parts) == 2:
+        try:
+            return int(parts[1]), None
+        except (ValueError, TypeError):
+            return None, None
+    return None, None
+
+
+async def wata_webhook(request: web.Request):
+    try:
+        ip = get_webhook_client_ip(request)
+        if await is_webhook_ip_blocked(ip):
+            return web.Response(status=429)
+
+        raw_json = await request.read()
+        try:
+            data = json.loads(raw_json)
+        except Exception as e:
+            logger.error(f"[WATA] Невалидный JSON в webhook: {e}")
+            return web.Response(status=400)
+
+        signature = request.headers.get("X-Signature")
+        if not signature:
+            logger.error("[WATA] Нет подписи X-Signature в заголовке")
+            await record_webhook_signature_failure(ip)
+            return web.Response(status=400)
+
+        try:
+            public_key_pem = await _get_wata_public_key()
+        except Exception as e:
+            logger.error(f"[WATA] Не удалось получить публичный ключ: {e}")
+            return web.Response(status=503)
+
+        if not await _verify_signature(raw_json, signature, public_key_pem):
+            logger.error("[WATA] Подпись не прошла проверку")
+            await record_webhook_signature_failure(ip)
+            return web.Response(status=400)
+
+        logger.info(f"[WATA] webhook: {json.dumps(data, ensure_ascii=False)}")
+
+        tx_status = data.get("transactionStatus")
+        order_id = str(data.get("orderId") or "")
+        transaction_id = str(data.get("transactionId") or "")
+        webhook_currency = str(data.get("currency") or "RUB").upper()
+        webhook_amount = data.get("amount")
+
+        if not order_id:
+            logger.error(f"[WATA] Пустой orderId в webhook: {data}")
+            return web.Response(status=400)
+
+        async with async_session_maker() as lookup_session:
+            pending = await get_payment_by_payment_id(lookup_session, order_id)
+
+        tg_id: int | None = None
+        rub_amount: float = 0.0
+        cassa_name: str | None = None
+
+        if pending:
+            try:
+                tg_id = int(pending["tg_id"])
+                rub_amount = float(pending["amount"])
+            except (TypeError, ValueError, KeyError):
+                pass
+            meta = pending.get("metadata") or {}
+            cassa_name = meta.get("cassa") if isinstance(meta, dict) else None
+        else:
+            fb_tg_id, fb_rub = _parse_wata_order_id(order_id)
+            if fb_tg_id is not None:
+                tg_id = fb_tg_id
+            if fb_rub and fb_rub > 0:
+                rub_amount = fb_rub
+
+        if tx_status == "Paid":
+            if tg_id is None:
+                logger.error(f"[WATA] Не удалось определить tg_id для orderId={order_id}")
+                return web.Response(status=400)
+
+            if rub_amount <= 0:
+                if webhook_currency == "RUB":
+                    try:
+                        raw_amount = float(webhook_amount or 0)
+                        raw_commission = float(data.get("commission") or 0)
+                        rub_amount = round(raw_amount - raw_commission, 2)
+                    except Exception:
+                        rub_amount = 0.0
+                if rub_amount <= 0:
+                    logger.error(f"[WATA] Не удалось определить RUB-сумму для зачисления: orderId={order_id}")
+                    return web.Response(status=400)
+
+            metadata_patch = {
+                "provider": _PROVIDER,
+                "wata_transaction_id": transaction_id or None,
+                "wata_currency": webhook_currency,
+                "wata_amount": webhook_amount,
+                "wata_commission": data.get("commission"),
+                "cassa": cassa_name,
+            }
+
+            update_currency: str | None = None
+            update_original_amount: float | None = None
+            if webhook_currency != "RUB":
+                update_currency = webhook_currency
+                try:
+                    update_original_amount = float(webhook_amount) if webhook_amount is not None else None
+                except (TypeError, ValueError):
+                    update_original_amount = None
+
+            parsed = ParsedPayment(
+                payment_id=order_id,
+                tg_id=int(tg_id),
+                amount=float(rub_amount),
+                currency="RUB",
+                metadata=metadata_patch,
+            )
+
+            result = await process_success_payment(
+                _PROVIDER,
+                parsed,
+                metadata_patch=metadata_patch,
+                update_currency=update_currency,
+                update_original_amount=update_original_amount,
+            )
+            if not result.ok:
+                logger.error(f"[WATA] Pipeline вернул ошибку: {result.error}, orderId={order_id}")
+                return web.Response(status=500)
+
+            logger.info(
+                f"[WATA] Платёж обработан: tg_id={tg_id}, amount={rub_amount:.2f} ₽, "
+                f"orderId={order_id}, transactionId={transaction_id}"
+            )
+            return web.Response(status=200, text="OK")
+
+        if tx_status == "Declined":
+            parsed_amount = float(rub_amount) if rub_amount > 0 else 0.0
+            if parsed_amount <= 0:
+                try:
+                    parsed_amount = float(webhook_amount or 0)
+                except Exception:
+                    parsed_amount = 0.0
+
+            parsed = ParsedPayment(
+                payment_id=order_id,
+                tg_id=int(tg_id) if tg_id is not None else None,
+                amount=parsed_amount,
+                currency=webhook_currency,
+            )
+            await process_cancelled_payment(_PROVIDER, parsed, new_status="failed")
+
+            logger.warning(f"[WATA] Транзакция отклонена: orderId={order_id}")
+            return web.Response(status=200, text="OK")
+
+        logger.warning(f"[WATA] Неизвестный статус транзакции: {tx_status}")
+        return web.Response(status=200, text="OK")
+    except Exception as e:
+        logger.error(f"[WATA] Ошибка в webhook: {e}", exc_info=True)
+        return web.Response(status=500)

web/__init__.py

@@ -2,16 +2,19 @@ from aiohttp.web_urldispatcher import UrlDispatcher
 
 from services.payments.heleket.webhook import heleket_webhook
 from services.payments.kassai.webhook import kassai_webhook
+from services.payments.wata.webhook import wata_webhook
 from utils.modules_loader import load_module_webhooks
 
 
 KASSAI_WEBHOOK_PATH = "/kassai/webhook"
 HELEKET_WEBHOOK_PATH = "/heleket/webhook"
+WATA_WEBHOOK_PATH = "/wata/webhook"
 
 
 async def register_web_routes(router: UrlDispatcher) -> None:
     router.add_post(KASSAI_WEBHOOK_PATH, kassai_webhook)
     router.add_post(HELEKET_WEBHOOK_PATH, heleket_webhook)
+    router.add_post(WATA_WEBHOOK_PATH, wata_webhook)
 
     try:
         module_webhooks = load_module_webhooks()