diff --git a/.env.example b/.env.example index 9f658e4a..8e3b9100 100644 --- a/.env.example +++ b/.env.example @@ -187,6 +187,16 @@ REMNAWAVE_AUTO_SYNC_ENABLED=false # Времена синхронизации (через запятую, формат HH:MM по МСК) REMNAWAVE_AUTO_SYNC_TIMES=03:00 +# ===== REMNAWAVE WEBHOOKS (входящие события из панели) ===== +# Включить приём вебхуков от панели Remnawave (real-time события) +REMNAWAVE_WEBHOOK_ENABLED=false +# Путь для приёма вебхуков (должен совпадать с настройкой в панели) +REMNAWAVE_WEBHOOK_PATH=/remnawave-webhook +# Общий секрет для подписи HMAC-SHA256 (минимум 32 символа) +# Сгенерируйте: openssl rand -hex 32 +# ВАЖНО: этот же секрет указывается в панели Remnawave при создании вебхука +REMNAWAVE_WEBHOOK_SECRET= + # Теги пользователей в Remnawave (A-Z, 0-9, _, макс. 16 символов) # Тег для пробных пользователей (опционально) # TRIAL_USER_TAG=TRIAL diff --git a/README.md b/README.md index a846714e..a9370f4c 100644 --- a/README.md +++ b/README.md @@ -610,6 +610,16 @@ hooks.domain.com { } } + handle /remnawave-webhook { + reverse_proxy remnawave_bot:8080 { + header_up Host {host} + header_up X-Real-IP {remote_host} + transport http { + read_buffer 0 + } + } + } + # app-config.json с CORS handle /app-config.json { header Access-Control-Allow-Origin "*" @@ -818,6 +828,18 @@ http { proxy_request_buffering off; } + location = /remnawave-webhook { + proxy_pass http://remnawave_bot_unified; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 120s; + proxy_send_timeout 120s; + proxy_buffering off; + proxy_request_buffering off; + } + # app-config.json с CORS location = /app-config.json { add_header Access-Control-Allow-Origin "*"; @@ -1055,6 +1077,103 @@ REMNAWAVE_SECRET_KEY=XXXXXXX:DDDDDDDD REMNAWAVE_SECRET_KEY=secret_key_name ``` +### 📡 Вебхуки Remnawave (real-time события) + +Бот может принимать входящие вебхуки от панели Remnawave для мгновенной реакции на события подписок. Это значительно улучшает скорость обновления данных по сравнению с периодической синхронизацией. + +#### Поддерживаемые события + +| Событие | Описание | +|---------|----------| +| `user.expired` | Подписка истекла | +| `user.disabled` | Подписка деактивирована | +| `user.enabled` | Подписка активирована | +| `user.limited` | Превышен лимит трафика | +| `user.traffic_reset` | Трафик сброшен | +| `user.modified` | Данные подписки изменены (трафик, дата, URL) | +| `user.deleted` | Пользователь удалён | +| `user.revoked` | Ключи подписки отозваны | +| `user.created` | Пользователь создан | +| `user.expires_in_*` | Предупреждения об истечении (72ч, 48ч, 24ч) | +| `user.first_connected` | Первое подключение | +| `user.bandwidth_usage_threshold_reached` | Порог трафика достигнут | +| `user_hwid_devices.*` | Устройство добавлено/удалено | +| `node.*`, `service.*` | Административные события (ноды, сервис) | + +#### Настройка + +**1. Переменные окружения в `.env`:** + +```env +REMNAWAVE_WEBHOOK_ENABLED=true +REMNAWAVE_WEBHOOK_PATH=/remnawave-webhook +REMNAWAVE_WEBHOOK_SECRET=your_secret_min_32_chars_here +``` + +Сгенерируйте секрет: + +```bash +openssl rand -hex 32 +``` + +**2. Настройка в панели Remnawave:** + +В панели Remnawave перейдите в раздел **Настройки > Вебхуки** и создайте новый вебхук: + +- **URL**: `https://hooks.domain.com/remnawave-webhook` +- **Secret**: тот же секрет, что и в `REMNAWAVE_WEBHOOK_SECRET` +- **Events**: выберите нужные события или все + +**3. Настройка прокси:** + +Добавьте путь `/remnawave-webhook` в конфигурацию обратного прокси. + +**Caddy:** + +```caddy +handle /remnawave-webhook { + reverse_proxy remnawave_bot:8080 { + header_up Host {host} + header_up X-Real-IP {remote_host} + transport http { + read_buffer 0 + } + } +} +``` + +**Nginx:** + +```nginx +location = /remnawave-webhook { + proxy_pass http://remnawave_bot_unified; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 120s; + proxy_send_timeout 120s; + proxy_buffering off; + proxy_request_buffering off; +} +``` + +**4. Проверка работоспособности:** + +```bash +# Health-check (GET запрос) +curl -s https://hooks.domain.com/remnawave-webhook | jq + +# Ожидаемый ответ: +# {"status": "ok", "service": "remnawave_webhook", "enabled": true} +``` + +**Важно:** +- Секрет должен быть не менее 32 символов +- Бот верифицирует подпись `X-Remnawave-Signature` (HMAC-SHA256) для каждого запроса +- При включённых вебхуках бот автоматически защищает подписки от перезаписи данными из периодической синхронизации в течение 60 секунд после получения события +- Если бот и панель на одном сервере, URL вебхука может быть `http://remnawave_bot:8080/remnawave-webhook` (внутри Docker-сети) + ### 💳 Freekassa Платёжный провайдер [Freekassa](https://freekassa.ru) поддерживает NSPK СБП и банковские карты. diff --git a/app/database/crud/subscription.py b/app/database/crud/subscription.py index 06dd6c64..c6239162 100644 --- a/app/database/crud/subscription.py +++ b/app/database/crud/subscription.py @@ -1,6 +1,6 @@ import logging from collections.abc import Iterable -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from typing import Optional from sqlalchemy import and_, delete, func, select @@ -23,6 +23,16 @@ from app.utils.timezone import format_local_datetime logger = logging.getLogger(__name__) +_WEBHOOK_GUARD_SECONDS = 60 + + +def is_recently_updated_by_webhook(subscription: Subscription) -> bool: + """Return True if subscription was updated by webhook within guard window.""" + if not subscription.last_webhook_update_at: + return False + elapsed = (datetime.now(UTC).replace(tzinfo=None) - subscription.last_webhook_update_at).total_seconds() + return elapsed < _WEBHOOK_GUARD_SECONDS + async def get_subscription_by_user_id(db: AsyncSession, user_id: int) -> Subscription | None: result = await db.execute( diff --git a/app/database/models.py b/app/database/models.py index 0eab896f..fc265fcf 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -1137,6 +1137,8 @@ class Subscription(Base): created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) + last_webhook_update_at = Column(DateTime, nullable=True) + remnawave_short_uuid = Column(String(255), nullable=True) # Тариф (для режима продаж "Тарифы") diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py index 48a7f179..abde8182 100644 --- a/app/database/universal_migration.py +++ b/app/database/universal_migration.py @@ -3613,6 +3613,33 @@ async def add_subscription_crypto_link_column() -> bool: return False +async def add_subscription_last_webhook_update_column() -> bool: + column_exists = await check_column_exists('subscriptions', 'last_webhook_update_at') + if column_exists: + logger.info('ℹ️ Колонка last_webhook_update_at уже существует') + return True + + try: + async with engine.begin() as conn: + db_type = await get_database_type() + + if db_type == 'sqlite': + await conn.execute(text('ALTER TABLE subscriptions ADD COLUMN last_webhook_update_at DATETIME')) + elif db_type == 'postgresql': + await conn.execute(text('ALTER TABLE subscriptions ADD COLUMN last_webhook_update_at TIMESTAMP')) + elif db_type == 'mysql': + await conn.execute(text('ALTER TABLE subscriptions ADD COLUMN last_webhook_update_at DATETIME')) + else: + logger.error(f'Неподдерживаемый тип БД для добавления last_webhook_update_at: {db_type}') + return False + + logger.info('✅ Добавлена колонка last_webhook_update_at в таблицу subscriptions') + return True + except Exception as e: + logger.error(f'Ошибка добавления колонки last_webhook_update_at: {e}') + return False + + async def fix_foreign_keys_for_user_deletion(): try: async with engine.begin() as conn: @@ -7104,6 +7131,13 @@ async def run_universal_migration(): else: logger.warning('⚠️ Проблемы с колонками OAuth провайдеров') + logger.info('=== ДОБАВЛЕНИЕ КОЛОНКИ LAST_WEBHOOK_UPDATE_AT ===') + webhook_column_ready = await add_subscription_last_webhook_update_column() + if webhook_column_ready: + logger.info('✅ Колонка last_webhook_update_at готова') + else: + logger.warning('⚠️ Проблемы с колонкой last_webhook_update_at') + async with engine.begin() as conn: total_subs = await conn.execute(text('SELECT COUNT(*) FROM subscriptions')) unique_users = await conn.execute(text('SELECT COUNT(DISTINCT user_id) FROM subscriptions')) diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py index 0e744716..a7f2aa6b 100644 --- a/app/services/monitoring_service.py +++ b/app/services/monitoring_service.py @@ -262,9 +262,18 @@ class MonitoringService: async def _check_expired_subscriptions(self, db: AsyncSession): try: + from app.database.crud.subscription import is_recently_updated_by_webhook + expired_subscriptions = await get_expired_subscriptions(db) for subscription in expired_subscriptions: + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск expire подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue + from app.database.crud.subscription import expire_subscription await expire_subscription(db, subscription) @@ -288,6 +297,15 @@ class MonitoringService: async def update_remnawave_user(self, db: AsyncSession, subscription: Subscription) -> RemnaWaveUser | None: try: + from app.database.crud.subscription import is_recently_updated_by_webhook + + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск RemnaWave обновления подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + return None + user = await get_user_by_id(db, subscription.user_id) if not user or not user.remnawave_uuid: logger.error(f'RemnaWave UUID не найден для пользователя {subscription.user_id}') @@ -299,6 +317,14 @@ class MonitoringService: except Exception: pass + # Re-check guard after refresh (webhook could have committed between first check and refresh) + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск RemnaWave обновления подписки %s: обновлена вебхуком недавно (после refresh)', + subscription.id, + ) + return None + current_time = datetime.utcnow() is_active = subscription.status == SubscriptionStatus.ACTIVE.value and subscription.end_date > current_time @@ -550,6 +576,8 @@ class MonitoringService: logger.error(f'Ошибка проверки неактивных тестовых подписок: {e}') async def _check_trial_channel_subscriptions(self, db: AsyncSession): + from app.database.crud.subscription import is_recently_updated_by_webhook + if not settings.CHANNEL_IS_REQUIRED_SUB: return @@ -636,6 +664,12 @@ class MonitoringService: continue if subscription.status == SubscriptionStatus.ACTIVE.value and subscription.is_trial and not is_member: + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск деактивации trial подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue subscription = await deactivate_subscription(db, subscription) disabled_count += 1 logger.info( @@ -670,6 +704,12 @@ class MonitoringService: 'trial_channel_unsubscribed', ) elif subscription.status == SubscriptionStatus.DISABLED.value and subscription.is_trial and is_member: + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск реактивации trial подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue subscription.status = SubscriptionStatus.ACTIVE.value subscription.updated_at = datetime.utcnow() await db.commit() @@ -1007,6 +1047,15 @@ class MonitoringService: failed_count = 0 for subscription in autopay_subscriptions: + from app.database.crud.subscription import is_recently_updated_by_webhook + + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск автоплатежа подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue + user = subscription.user if not user: continue @@ -1874,11 +1923,19 @@ class MonitoringService: } async def force_check_subscriptions(self, db: AsyncSession) -> dict[str, int]: + from app.database.crud.subscription import is_recently_updated_by_webhook + try: expired_subscriptions = await get_expired_subscriptions(db) expired_count = 0 for subscription in expired_subscriptions: + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск force-check подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue await deactivate_subscription(db, subscription) expired_count += 1 diff --git a/app/services/remnawave_service.py b/app/services/remnawave_service.py index 59c7d7f0..36d35ad3 100644 --- a/app/services/remnawave_service.py +++ b/app/services/remnawave_service.py @@ -1454,10 +1454,20 @@ class RemnaWaveService: for telegram_id, db_user in users_to_deactivate: cleanup_mutation: _UUIDMapMutation | None = None try: - logger.info(f'🗑️ Деактивация подписки пользователя {telegram_id} (нет в панели)') - subscription = db_user.subscription + # Skip if recently updated by webhook + from app.database.crud.subscription import is_recently_updated_by_webhook + + if subscription and is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск деактивации подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + continue + + logger.info(f'🗑️ Деактивация подписки пользователя {telegram_id} (нет в панели)') + if db_user.remnawave_uuid and hwid_api_client: try: devices_reset = await hwid_api_client.reset_user_devices(db_user.remnawave_uuid) @@ -1664,7 +1674,7 @@ class RemnaWaveService: async def _update_subscription_from_panel_data(self, db: AsyncSession, user, panel_user): try: - from app.database.crud.subscription import get_subscription_by_user_id + from app.database.crud.subscription import get_subscription_by_user_id, is_recently_updated_by_webhook from app.database.models import SubscriptionStatus # Всегда используем async CRUD запрос для получения подписки, @@ -1675,6 +1685,14 @@ class RemnaWaveService: await self._create_subscription_from_panel_data(db, user, panel_user) return + # Skip if recently updated by webhook (prevent stale data overwrite) + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск синхронизации подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + return + panel_status = panel_user.get('status', 'ACTIVE') expire_at_str = panel_user.get('expireAt', '') @@ -2499,12 +2517,20 @@ class RemnaWaveService: await self._update_subscription_from_panel_data(db, user, panel_user) stats['updated'] += 1 elif subscription.status != SubscriptionStatus.DISABLED.value: - logger.info(f'🗑️ Деактивируем подписку пользователя {user.telegram_id} (нет в панели)') + from app.database.crud.subscription import ( + deactivate_subscription, + is_recently_updated_by_webhook, + ) - from app.database.crud.subscription import deactivate_subscription - - await deactivate_subscription(db, subscription) - stats['updated'] += 1 + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Пропуск деактивации подписки %s: обновлена вебхуком недавно', + subscription.id, + ) + else: + logger.info(f'🗑️ Деактивируем подписку пользователя {user.telegram_id} (нет в панели)') + await deactivate_subscription(db, subscription) + stats['updated'] += 1 except Exception as sub_error: logger.error(f'❌ Ошибка синхронизации подписки {subscription.id}: {sub_error}') @@ -2547,6 +2573,8 @@ class RemnaWaveService: user = subscription.user issues_fixed = 0 + from app.database.crud.subscription import is_recently_updated_by_webhook + current_time = self._now_utc() # Конвертируем end_date в UTC для корректного сравнения end_date_utc = self._local_to_utc(subscription.end_date) @@ -2555,6 +2583,7 @@ class RemnaWaveService: if ( end_date_utc + expiry_buffer <= current_time and subscription.status == SubscriptionStatus.ACTIVE.value + and not is_recently_updated_by_webhook(subscription) ): time_since_expiry = current_time - end_date_utc logger.warning( diff --git a/app/services/remnawave_webhook_service.py b/app/services/remnawave_webhook_service.py index bdec7818..1e22512a 100644 --- a/app/services/remnawave_webhook_service.py +++ b/app/services/remnawave_webhook_service.py @@ -392,6 +392,15 @@ class RemnaWaveWebhookService: except Exception: logger.exception('Notification delivery failed for user %s, text_key %s', user.id, text_key) + # ------------------------------------------------------------------ + # Webhook timestamp helper + # ------------------------------------------------------------------ + + @staticmethod + def _stamp_webhook_update(subscription: Subscription) -> None: + """Mark subscription as recently updated by webhook to prevent sync overwrite.""" + subscription.last_webhook_update_at = datetime.now(UTC).replace(tzinfo=None) + # ------------------------------------------------------------------ # User event handlers # ------------------------------------------------------------------ @@ -399,36 +408,52 @@ class RemnaWaveWebhookService: async def _handle_user_expired( self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: - if subscription and subscription.status != SubscriptionStatus.EXPIRED.value: - await expire_subscription(db, subscription) - logger.info('Webhook: subscription %s expired for user %s', subscription.id, user.id) + if subscription: + self._stamp_webhook_update(subscription) + if subscription.status != SubscriptionStatus.EXPIRED.value: + await expire_subscription(db, subscription) + logger.info('Webhook: subscription %s expired for user %s', subscription.id, user.id) + else: + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_EXPIRED', reply_markup=self._get_renew_keyboard(user)) async def _handle_user_disabled( self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: - if subscription and subscription.status != SubscriptionStatus.DISABLED.value: - await deactivate_subscription(db, subscription) - logger.info('Webhook: subscription %s disabled for user %s', subscription.id, user.id) + if subscription: + self._stamp_webhook_update(subscription) + if subscription.status != SubscriptionStatus.DISABLED.value: + await deactivate_subscription(db, subscription) + logger.info('Webhook: subscription %s disabled for user %s', subscription.id, user.id) + else: + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_DISABLED', reply_markup=self._get_subscription_keyboard(user)) async def _handle_user_enabled( self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: - if subscription and subscription.status == SubscriptionStatus.DISABLED.value: - await reactivate_subscription(db, subscription) - logger.info('Webhook: subscription %s re-enabled for user %s', subscription.id, user.id) + if subscription: + self._stamp_webhook_update(subscription) + if subscription.status == SubscriptionStatus.DISABLED.value: + await reactivate_subscription(db, subscription) + logger.info('Webhook: subscription %s re-enabled for user %s', subscription.id, user.id) + else: + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_ENABLED', reply_markup=self._get_connect_keyboard(user)) async def _handle_user_limited( self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: - if subscription and subscription.status == SubscriptionStatus.ACTIVE.value: - await deactivate_subscription(db, subscription) - logger.info('Webhook: subscription %s limited (traffic) for user %s', subscription.id, user.id) + if subscription: + self._stamp_webhook_update(subscription) + if subscription.status == SubscriptionStatus.ACTIVE.value: + await deactivate_subscription(db, subscription) + logger.info('Webhook: subscription %s limited (traffic) for user %s', subscription.id, user.id) + else: + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_LIMITED', reply_markup=self._get_traffic_keyboard(user)) @@ -436,6 +461,7 @@ class RemnaWaveWebhookService: self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: if subscription: + self._stamp_webhook_update(subscription) await update_subscription_usage(db, subscription, 0.0) # Re-enable if was disabled due to traffic limit if subscription.status == SubscriptionStatus.DISABLED.value: @@ -496,24 +522,32 @@ class RemnaWaveWebhookService: subscription.subscription_url = subscription_url changed = True + # Always stamp to protect from sync overwrite, even if no fields changed + self._stamp_webhook_update(subscription) if changed: subscription.updated_at = datetime.now(UTC).replace(tzinfo=None) - await db.flush() logger.info('Webhook: subscription %s modified (synced from panel) for user %s', subscription.id, user.id) + await db.commit() async def _handle_user_deleted( self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict ) -> None: - if subscription and subscription.status != SubscriptionStatus.EXPIRED.value: - await expire_subscription(db, subscription) - logger.info( - 'Webhook: subscription %s marked expired (user deleted in panel) for user %s', subscription.id, user.id - ) + if subscription: + self._stamp_webhook_update(subscription) + if subscription.status != SubscriptionStatus.EXPIRED.value: + await expire_subscription(db, subscription) + logger.info( + 'Webhook: subscription %s marked expired (user deleted in panel) for user %s', + subscription.id, + user.id, + ) + else: + await db.commit() # Clear remnawave linkage if user.remnawave_uuid: user.remnawave_uuid = None - await db.flush() + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_DELETED', reply_markup=self._get_renew_keyboard(user)) @@ -536,12 +570,14 @@ class RemnaWaveWebhookService: subscription.subscription_crypto_link = new_crypto_link changed = True + # Always stamp to protect from sync overwrite + self._stamp_webhook_update(subscription) if changed: subscription.updated_at = datetime.now(UTC).replace(tzinfo=None) - await db.flush() logger.info( 'Webhook: subscription %s credentials revoked/updated for user %s', subscription.id, user.id ) + await db.commit() await self._notify_user(user, 'WEBHOOK_SUB_REVOKED', reply_markup=self._get_connect_keyboard(user)) diff --git a/tests/external/test_yookassa_webhook.py b/tests/external/test_yookassa_webhook.py index 203b9430..6b8d99fa 100644 --- a/tests/external/test_yookassa_webhook.py +++ b/tests/external/test_yookassa_webhook.py @@ -111,10 +111,19 @@ async def _post_webhook(client: TestClient, payload: dict, **headers: str) -> we def _patch_get_db(monkeypatch: pytest.MonkeyPatch) -> None: - async def fake_get_db(): - yield DummyDB() + """Mock AsyncSessionLocal used by the webhook handler.""" + from unittest.mock import MagicMock - monkeypatch.setattr('app.external.yookassa_webhook.get_db', fake_get_db) + mock_session = AsyncMock() + mock_session.commit = AsyncMock() + mock_session.rollback = AsyncMock() + mock_session.execute = AsyncMock() + + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=mock_session) + ctx.__aexit__ = AsyncMock(return_value=False) + + monkeypatch.setattr('app.external.yookassa_webhook.AsyncSessionLocal', lambda: ctx) @pytest.mark.asyncio