fix: repair missing panel user references

backend/bot/services/tariff_worker.py

@@ -155,6 +155,13 @@ class TariffTrafficWorker:
                 cached_panel_user = panel_users_by_uuid.get(str(sub.panel_user_uuid))
                 if cached_panel_user is not None:
                     return cached_panel_user
+                return await self._repair_missing_panel_user_for_subscription(
+                    session,
+                    sub,
+                    panel_users_by_uuid=panel_users_by_uuid,
+                    semaphore=semaphore,
+                    confirmed_missing=True,
+                )
 
             async with semaphore:
                 try:
@@ -167,12 +174,22 @@ class TariffTrafficWorker:
                         sub.panel_user_uuid,
                     )
                     return {}
-            return data or {}
+            if data:
+                return data
+            return await self._repair_missing_panel_user_for_subscription(
+                session,
+                sub,
+                panel_users_by_uuid=None,
+                semaphore=semaphore,
+                confirmed_missing=False,
+            )
 
         for chunk_start in range(0, len(subs), TARIFF_WORKER_BATCH_SIZE):
             chunk = subs[chunk_start : chunk_start + TARIFF_WORKER_BATCH_SIZE]
             panel_payloads = await asyncio.gather(*(_fetch_panel(s) for s in chunk))
             for sub, panel_data in zip(chunk, panel_payloads):
+                if not panel_data:
+                    continue
                 try:
                     tariff = self.settings.tariffs_config.require(sub.tariff_key)
                 except Exception:
@@ -255,6 +272,69 @@ class TariffTrafficWorker:
         )
         return by_uuid
 
+    async def _repair_missing_panel_user_for_subscription(
+        self,
+        session: AsyncSession,
+        sub: Subscription,
+        *,
+        panel_users_by_uuid: Optional[dict[str, dict]],
+        semaphore: asyncio.Semaphore,
+        confirmed_missing: bool,
+    ) -> dict:
+        current_uuid = str(getattr(sub, "panel_user_uuid", "") or "").strip()
+        try:
+            user_id = int(sub.user_id)
+        except (TypeError, ValueError):
+            user_id = 0
+        db_user = await user_dal.get_user_by_id(session, user_id) if user_id else None
+        canonical_uuid = str(getattr(db_user, "panel_user_uuid", "") or "").strip()
+
+        if canonical_uuid and canonical_uuid != current_uuid:
+            panel_user = None
+            if panel_users_by_uuid is not None:
+                panel_user = panel_users_by_uuid.get(canonical_uuid)
+            else:
+                async with semaphore:
+                    try:
+                        panel_user = await self.panel_service.get_user_by_uuid(
+                            canonical_uuid,
+                            log_response=False,
+                        )
+                    except Exception:
+                        logging.exception(
+                            "TariffTrafficWorker: failed to fetch canonical panel user %s",
+                            canonical_uuid,
+                        )
+                        panel_user = None
+            if panel_user:
+                logging.warning(
+                    "TariffTrafficWorker: repaired subscription %s panel UUID %s -> %s",
+                    sub.subscription_id,
+                    current_uuid,
+                    canonical_uuid,
+                )
+                sub.panel_user_uuid = canonical_uuid
+                return panel_user
+
+        if confirmed_missing:
+            sub.is_active = False
+            sub.skip_notifications = True
+            sub.status_from_panel = "PANEL_USER_NOT_FOUND"
+            logging.warning(
+                "TariffTrafficWorker: deactivated subscription %s because panel user %s "
+                "is missing",
+                sub.subscription_id,
+                current_uuid,
+            )
+        else:
+            logging.warning(
+                "TariffTrafficWorker: skipping subscription %s because panel user %s "
+                "could not be fetched",
+                sub.subscription_id,
+                current_uuid,
+            )
+        return {}
+
     async def _ensure_period_reset_strategy(
         self,
         sub: Subscription,

tests/test_tariff_worker.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import tempfile
 import unittest
@@ -617,3 +618,107 @@ class TariffWorkerTests(unittest.IsolatedAsyncioTestCase):
 
         self.assertIsNone(result)
         panel_service.get_all_panel_users.assert_not_awaited()
+
+    async def test_missing_panel_subscription_repairs_to_user_panel_uuid(self):
+        panel_service = AsyncMock(spec=PanelApiService)
+        worker = TariffTrafficWorker(
+            settings=SimpleNamespace(),
+            session_factory=SimpleNamespace(),
+            panel_service=panel_service,
+            subscription_service=SimpleNamespace(),
+        )
+        sub = SimpleNamespace(
+            subscription_id=10,
+            user_id=123,
+            panel_user_uuid="old-panel",
+            is_active=True,
+            status_from_panel="ACTIVE",
+            skip_notifications=False,
+        )
+        panel_user = {"uuid": "new-panel", "username": "tg_123"}
+
+        with patch(
+            "bot.services.tariff_worker.user_dal.get_user_by_id",
+            new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="new-panel")),
+        ):
+            result = await worker._repair_missing_panel_user_for_subscription(
+                AsyncMock(),
+                sub,
+                panel_users_by_uuid={"new-panel": panel_user},
+                semaphore=asyncio.Semaphore(1),
+                confirmed_missing=True,
+            )
+
+        self.assertEqual(result, panel_user)
+        self.assertEqual(sub.panel_user_uuid, "new-panel")
+        self.assertTrue(sub.is_active)
+        panel_service.get_user_by_uuid.assert_not_awaited()
+
+    async def test_missing_panel_subscription_deactivates_when_bulk_prefetch_confirms_absent(self):
+        panel_service = AsyncMock(spec=PanelApiService)
+        worker = TariffTrafficWorker(
+            settings=SimpleNamespace(),
+            session_factory=SimpleNamespace(),
+            panel_service=panel_service,
+            subscription_service=SimpleNamespace(),
+        )
+        sub = SimpleNamespace(
+            subscription_id=11,
+            user_id=123,
+            panel_user_uuid="missing-panel",
+            is_active=True,
+            status_from_panel="ACTIVE",
+            skip_notifications=False,
+        )
+
+        with patch(
+            "bot.services.tariff_worker.user_dal.get_user_by_id",
+            new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="missing-panel")),
+        ):
+            result = await worker._repair_missing_panel_user_for_subscription(
+                AsyncMock(),
+                sub,
+                panel_users_by_uuid={},
+                semaphore=asyncio.Semaphore(1),
+                confirmed_missing=True,
+            )
+
+        self.assertEqual(result, {})
+        self.assertFalse(sub.is_active)
+        self.assertTrue(sub.skip_notifications)
+        self.assertEqual(sub.status_from_panel, "PANEL_USER_NOT_FOUND")
+
+    async def test_missing_panel_subscription_only_skips_when_absence_is_not_confirmed(self):
+        panel_service = AsyncMock(spec=PanelApiService)
+        panel_service.get_user_by_uuid = AsyncMock(return_value=None)
+        worker = TariffTrafficWorker(
+            settings=SimpleNamespace(),
+            session_factory=SimpleNamespace(),
+            panel_service=panel_service,
+            subscription_service=SimpleNamespace(),
+        )
+        sub = SimpleNamespace(
+            subscription_id=12,
+            user_id=123,
+            panel_user_uuid="missing-panel",
+            is_active=True,
+            status_from_panel="ACTIVE",
+            skip_notifications=False,
+        )
+
+        with patch(
+            "bot.services.tariff_worker.user_dal.get_user_by_id",
+            new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="missing-panel")),
+        ):
+            result = await worker._repair_missing_panel_user_for_subscription(
+                AsyncMock(),
+                sub,
+                panel_users_by_uuid=None,
+                semaphore=asyncio.Semaphore(1),
+                confirmed_missing=False,
+            )
+
+        self.assertEqual(result, {})
+        self.assertTrue(sub.is_active)
+        self.assertFalse(sub.skip_notifications)
+        self.assertEqual(sub.status_from_panel, "ACTIVE")