diff --git a/backend/bot/handlers/admin/common.py b/backend/bot/handlers/admin/common.py index 19794c3..87e53b5 100644 --- a/backend/bot/handlers/admin/common.py +++ b/backend/bot/handlers/admin/common.py @@ -156,7 +156,6 @@ async def admin_panel_actions_callback_handler( panel_service=panel_service, session=session, ) - await callback.answer(_("admin_sync_initiated_from_panel")) elif action == "queue_status": await show_queue_status_handler(callback, i18n_data) elif action == "view_payments": diff --git a/backend/bot/handlers/admin/sync_admin.py b/backend/bot/handlers/admin/sync_admin.py index 0baad9e..261b724 100644 --- a/backend/bot/handlers/admin/sync_admin.py +++ b/backend/bot/handlers/admin/sync_admin.py @@ -9,8 +9,8 @@ from aiogram.filters import Command from sqlalchemy import func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession +from bot.infra.webhook_queue import enqueue_webhook_event from bot.middlewares.i18n import JsonI18n -from bot.services.notification_service import NotificationService from bot.services.panel_api_service import PanelApiService from bot.utils.text_sanitizer import panel_description_from_profile from config.settings import Settings @@ -1613,63 +1613,74 @@ async def sync_command_handler( return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) - target_chat_id = ( - message_event.chat.id - if isinstance(message_event, types.Message) - else (message_event.message.chat.id if message_event.message else None) - ) + target_chat_id = _sync_request_target_chat_id(message_event) if not target_chat_id: logging.error("Sync handler: could not determine target_chat_id.") if isinstance(message_event, types.CallbackQuery): await message_event.answer("Error initiating sync.", show_alert=True) return - if isinstance(message_event, types.Message): - await message_event.answer(_("sync_started_simple")) + requested_by = getattr(getattr(message_event, "from_user", None), "id", None) + queued = await _enqueue_manual_panel_sync( + settings, + requested_by=requested_by, + target_chat_id=target_chat_id, + language=current_lang, + ) + if not queued: + logging.warning("Admin (%s) failed to enqueue manual panel sync.", requested_by) + await _answer_sync_request(message_event, _("sync_failed_simple"), show_alert=True) + return - logging.info(f"Admin ({message_event.from_user.id}) triggered panel sync.") + await _answer_sync_request( + message_event, + _("admin_sync_initiated_from_panel") + if isinstance(message_event, types.CallbackQuery) + else _("sync_started_simple"), + ) + logging.info("Admin (%s) queued panel sync from bot.", requested_by) - # Use the extracted perform_sync function - try: - sync_result = await perform_sync(panel_service, session, settings, i18n) - status = sync_result.get("status") - details = sync_result.get("details", "No details available") - errors = sync_result.get("errors", []) +def _sync_request_target_chat_id(message_event: Union[types.Message, types.CallbackQuery]): + chat = getattr(message_event, "chat", None) + if chat and getattr(chat, "id", None) is not None: + return chat.id + callback_message = getattr(message_event, "message", None) + callback_chat = getattr(callback_message, "chat", None) + if callback_chat and getattr(callback_chat, "id", None) is not None: + return callback_chat.id + return None - # Simple confirmation message to admin - if status == "failed": - await bot.send_message(target_chat_id, _("sync_failed_simple")) - elif status == "completed_with_errors": - await bot.send_message( - target_chat_id, - _("sync_errors_simple", errors_count=len(errors)), - ) - else: - await bot.send_message(target_chat_id, _("sync_success_simple")) - # Send notification to log channel with proper thread handling - try: - notification_service = NotificationService(bot, settings, i18n) - await notification_service.notify_panel_sync( - status, - details, - sync_result.get("users_processed", 0), - sync_result.get("subs_synced", 0), - ) - except Exception as e_notification: - logging.error(f"Failed to send sync notification: {e_notification}") +async def _answer_sync_request( + message_event: Union[types.Message, types.CallbackQuery], + text: str, + *, + show_alert: bool = False, +) -> None: + answer = getattr(message_event, "answer", None) + if not callable(answer): + return + if isinstance(message_event, types.CallbackQuery): + await answer(text, show_alert=show_alert) + return + await answer(text) - except Exception as e_sync_global: - logging.error(f"Global error during /sync command: {e_sync_global}", exc_info=True) - await bot.send_message(target_chat_id, _("sync_critical_error")) - # Send notification to log channel about failure - try: - notification_service = NotificationService(bot, settings, i18n) - await notification_service.notify_panel_sync("failed", str(e_sync_global), 0, 0) - except Exception as e_notification: - logging.error(f"Failed to send sync failure notification: {e_notification}") +async def _enqueue_manual_panel_sync( + settings: Settings, + *, + requested_by: Optional[int], + target_chat_id: int, + language: str, +) -> bool: + payload = { + "source": "bot_admin", + "requested_by": requested_by, + "target_chat_id": target_chat_id, + "language": language, + } + return await enqueue_webhook_event(settings, "panel_sync", payload, event_id=None) @router.message(Command("syncstatus")) diff --git a/backend/main_worker.py b/backend/main_worker.py index 075a68e..af303ee 100644 --- a/backend/main_worker.py +++ b/backend/main_worker.py @@ -88,6 +88,7 @@ async def _webhook_consumer(settings, session_factory, bot, i18n, services): payload.get("user") or {}, ) elif provider == "panel_sync": + sync_result = None async with redis_lock( settings, "panel-sync", @@ -95,12 +96,24 @@ async def _webhook_consumer(settings, session_factory, bot, i18n, services): ) as acquired: if acquired: async with session_factory() as session: - await perform_sync( + sync_result = await perform_sync( panel_service=services["panel_service"], session=session, settings=settings, i18n_instance=i18n, ) + else: + logging.info( + "Queued panel sync skipped because another sync holds the lock" + ) + if sync_result is not None: + await _notify_queued_panel_sync_result( + bot, + settings, + i18n, + payload, + sync_result, + ) else: logging.warning("Unknown webhook event provider: %s", provider) except Exception: @@ -115,6 +128,28 @@ async def _webhook_consumer(settings, session_factory, bot, i18n, services): ) +async def _notify_queued_panel_sync_result(bot, settings, i18n, payload, sync_result): + status = sync_result.get("status") + errors = sync_result.get("errors", []) + lang = payload.get("language") or settings.DEFAULT_LANGUAGE + _ = lambda key, **kwargs: i18n.gettext(lang, key, **kwargs) + + target_chat_id = payload.get("target_chat_id") + if target_chat_id: + try: + if status == "failed": + await bot.send_message(target_chat_id, _("sync_failed_simple")) + elif status == "completed_with_errors": + await bot.send_message( + target_chat_id, + _("sync_errors_simple", errors_count=len(errors)), + ) + else: + await bot.send_message(target_chat_id, _("sync_success_simple")) + except Exception: + logging.exception("Failed to send queued panel sync result to admin") + + async def _panel_sync_loop(settings, session_factory, i18n, services): while True: try: diff --git a/tests/test_admin_bot_sync_queue.py b/tests/test_admin_bot_sync_queue.py new file mode 100644 index 0000000..371bea5 --- /dev/null +++ b/tests/test_admin_bot_sync_queue.py @@ -0,0 +1,89 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from bot.handlers.admin import common as admin_common +from bot.handlers.admin import sync_admin as sync_module + + +class _I18n: + def gettext(self, _lang, key, **kwargs): + if kwargs: + return f"{key}:{kwargs}" + return key + + +def _settings(): + return SimpleNamespace(DEFAULT_LANGUAGE="ru") + + +class AdminBotSyncQueueTests(unittest.IsolatedAsyncioTestCase): + async def test_sync_command_enqueues_panel_sync_instead_of_running_inline(self): + enqueued = [] + + async def fake_enqueue(settings, provider, payload, *, event_id=None): + enqueued.append((settings, provider, payload, event_id)) + return True + + message = SimpleNamespace( + from_user=SimpleNamespace(id=42), + chat=SimpleNamespace(id=100500), + answer=AsyncMock(), + ) + i18n_data = {"current_language": "ru", "i18n_instance": _I18n()} + + with ( + patch.object(sync_module, "enqueue_webhook_event", fake_enqueue), + patch.object(sync_module, "perform_sync", AsyncMock()) as perform_sync, + ): + await sync_module.sync_command_handler( + message_event=message, + bot=AsyncMock(), + settings=_settings(), + i18n_data=i18n_data, + panel_service=AsyncMock(), + session=AsyncMock(), + ) + + perform_sync.assert_not_awaited() + message.answer.assert_awaited_once_with("sync_started_simple") + self.assertEqual(len(enqueued), 1) + _, provider, payload, event_id = enqueued[0] + self.assertEqual(provider, "panel_sync") + self.assertIsNone(event_id) + self.assertEqual( + payload, + { + "source": "bot_admin", + "requested_by": 42, + "target_chat_id": 100500, + "language": "ru", + }, + ) + + async def test_admin_sync_button_delegates_to_sync_handler_without_second_answer(self): + callback = SimpleNamespace( + data="admin_action:sync_panel", + from_user=SimpleNamespace(id=42), + message=SimpleNamespace(chat=SimpleNamespace(id=100500)), + answer=AsyncMock(), + ) + + with patch.object( + admin_common.admin_sync_handlers, + "sync_command_handler", + AsyncMock(), + ) as sync_command: + await admin_common.admin_panel_actions_callback_handler( + callback=callback, + state=AsyncMock(), + settings=_settings(), + i18n_data={"current_language": "ru", "i18n_instance": _I18n()}, + bot=AsyncMock(), + panel_service=AsyncMock(), + subscription_service=AsyncMock(), + session=AsyncMock(), + ) + + sync_command.assert_awaited_once() + callback.answer.assert_not_awaited()