Add priority-only Discord notifications

tests/test_discord_notifications.py

@@ -0,0 +1,149 @@
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+from aiohttp.test_utils import TestClient, TestServer
+
+from web.auth import AuthStore
+from web.discord import DiscordNotifier, validate_webhook_url
+from web.server import create_app
+
+
+WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/test_token"
+
+
+def campaign(game_name: str = "Priority Game", drop_id: str = "drop-1") -> SimpleNamespace:
+    game = SimpleNamespace(name=game_name)
+    benefit = SimpleNamespace(
+        name="Reward One",
+        image_url="https://static-cdn.jtvnw.net/reward.png",
+    )
+    drop = SimpleNamespace(
+        id=drop_id,
+        name="Reward Drop",
+        benefits=[benefit],
+        required_minutes=60,
+        ends_at=datetime.now(timezone.utc) + timedelta(days=1),
+        is_claimed=False,
+    )
+    item = SimpleNamespace(
+        id=f"campaign-{drop_id}",
+        name="Priority Campaign",
+        game=game,
+        drops=[drop],
+        image_url="https://static-cdn.jtvnw.net/category.jpg",
+        link_url="https://www.twitch.tv/drops/campaigns",
+        ends_at=datetime.now(timezone.utc) + timedelta(days=1),
+        claimed_drops=1,
+        total_drops=2,
+        finished=False,
+        can_earn_within=lambda _: True,
+    )
+    drop.campaign = item
+    return item
+
+
+class DiscordNotificationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = AuthStore(Path(self.temp.name, "auth.sqlite3"))
+        self.notifier = DiscordNotifier(self.store)
+        self.notifier.update({"webhook_url": WEBHOOK, "enabled": True})
+        self.payloads = []
+        self.notifier._schedule = self.payloads.append  # type: ignore[method-assign]
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def test_url_is_restricted_and_secret_is_not_returned(self) -> None:
+        self.assertEqual(validate_webhook_url(WEBHOOK), WEBHOOK)
+        with self.assertRaises(ValueError):
+            validate_webhook_url("https://discord.com.evil.example/api/webhooks/1/token")
+        with self.assertRaises(ValueError):
+            validate_webhook_url(f"{WEBHOOK}?redirect=https://example.com")
+        snapshot = self.notifier.snapshot()
+        self.assertTrue(snapshot["configured"])
+        self.assertNotIn("url", snapshot)
+        self.assertNotIn("test_token", str(snapshot))
+        self.assertTrue(DiscordNotifier(AuthStore(self.store.path)).snapshot()["configured"])
+
+    def test_priority_drop_messages_include_category_and_reward_images(self) -> None:
+        self.notifier.finish_inventory()
+        ignored = campaign("Ignored Game", "ignored")
+        selected = campaign()
+
+        self.notifier.observe_campaign(ignored, ["Priority Game"])
+        self.notifier.observe_campaign(selected, ["Priority Game"])
+        self.assertEqual(len(self.payloads), 1)
+        message = self.payloads[0]
+        self.assertEqual(message["allowed_mentions"], {"parse": []})
+        self.assertEqual(message["embeds"][0]["image"]["url"], selected.image_url)
+        self.assertEqual(
+            message["embeds"][1]["thumbnail"]["url"],
+            selected.drops[0].benefits[0].image_url,
+        )
+
+        selected.drops[0].is_claimed = True
+        self.notifier.drop_updated(selected.drops[0], ["Priority Game"])
+        self.notifier.drop_updated(selected.drops[0], ["Priority Game"])
+        self.assertEqual([payload["embeds"][0]["title"] for payload in self.payloads], [
+            "New priority Drops detected",
+            "Drop claimed",
+        ])
+
+    def test_idle_reason_is_deduplicated_and_recovery_is_reported(self) -> None:
+        selected = campaign()
+        channel = SimpleNamespace(
+            id=7,
+            name="Streamer",
+            game=selected.game,
+            online=True,
+            drops_enabled=False,
+        )
+        manager = SimpleNamespace(
+            _twitch=SimpleNamespace(
+                settings=SimpleNamespace(priority=["Priority Game"]),
+                channels={channel.id: channel},
+            ),
+            inv=SimpleNamespace(campaigns={selected.id: selected}),
+        )
+
+        self.notifier.idle(manager)
+        self.notifier.idle(manager)
+        channel.drops_enabled = True
+        self.notifier.watching(manager, channel)
+
+        self.assertEqual([payload["embeds"][0]["title"] for payload in self.payloads], [
+            "Drops disabled on live channels",
+            "Mining resumed",
+        ])
+
+
+class DiscordNotificationApiTests(unittest.IsolatedAsyncioTestCase):
+    async def test_authenticated_update_never_returns_the_webhook_token(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            auth_path = Path(directory, "auth.sqlite3")
+            AuthStore(auth_path).provision(
+                "correct horse battery", "recovery-code-long-enough"
+            )
+            app = create_app(auth_path, Path(directory), auto_start=False)
+            async with TestClient(TestServer(app)) as client:
+                login = await client.post(
+                    "/api/login", json={"password": "correct horse battery"}
+                )
+                csrf = (await login.json())["csrf_token"]
+                response = await client.put(
+                    "/api/notifications",
+                    headers={"X-CSRF-Token": csrf},
+                    json={"webhook_url": WEBHOOK, "enabled": True},
+                )
+                body = await response.json()
+
+                self.assertEqual(response.status, 200)
+                self.assertNotIn("test_token", str(body))
+
+
+if __name__ == "__main__":
+    unittest.main()

web/auth.py

@@ -113,6 +113,14 @@ def _initialize(self) -> None:
                     csrf_token TEXT NOT NULL,
                     expires_at INTEGER NOT NULL
                 );
+                CREATE TABLE IF NOT EXISTS settings (
+                    name TEXT PRIMARY KEY,
+                    value TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS notification_events (
+                    event_key TEXT PRIMARY KEY,
+                    notified_at INTEGER NOT NULL
+                );
                 """
             )
         if os.name != "nt":
@@ -224,3 +232,36 @@ def force_password(self, new_password: str) -> str:
             )
             db.execute("DELETE FROM sessions")
         return next_recovery
+
+    def get_settings(self, defaults: dict[str, str]) -> dict[str, str]:
+        with self._db() as db:
+            rows = db.execute("SELECT name, value FROM settings").fetchall()
+        return {**defaults, **dict(rows)}
+
+    def update_settings(self, values: dict[str, str | None]) -> None:
+        with self._db() as db:
+            for name, value in values.items():
+                if value is None:
+                    db.execute("DELETE FROM settings WHERE name = ?", (name,))
+                else:
+                    db.execute(
+                        "INSERT INTO settings(name, value) VALUES (?, ?) "
+                        "ON CONFLICT(name) DO UPDATE SET value = excluded.value",
+                        (name, value),
+                    )
+
+    def claim_notification_event(self, event_key: str, cooldown: int | None = None) -> bool:
+        now = int(time.time())
+        with self._db() as db:
+            row = db.execute(
+                "SELECT notified_at FROM notification_events WHERE event_key = ?",
+                (event_key,),
+            ).fetchone()
+            if row is not None and (cooldown is None or row[0] > now - cooldown):
+                return False
+            db.execute(
+                "INSERT INTO notification_events(event_key, notified_at) VALUES (?, ?) "
+                "ON CONFLICT(event_key) DO UPDATE SET notified_at = excluded.notified_at",
+                (event_key, now),
+            )
+        return True

web/controller.py

@@ -13,11 +13,13 @@
 from core.translate import _
 from core.utils import lock_file
 from network.twitch import Twitch
+from web.discord import DiscordNotifier
 from web.manager import WebManager
 
 
 class MinerController:
-    def __init__(self) -> None:
+    def __init__(self, notifier: DiscordNotifier | None = None) -> None:
+        self.notifier = notifier
         self.manager: WebManager | None = None
         self._client: Twitch | None = None
         self._task: asyncio.Task[None] | None = None
@@ -39,11 +41,13 @@ async def start(self) -> bool:
             await asyncio.sleep(0)
             return True
 
-    async def stop(self) -> bool:
+    async def stop(self, *, notify: bool = True) -> bool:
         async with self._lock:
             if not self.running or self.manager is None:
                 return False
             task = self._task
+            if notify and self.notifier is not None:
+                self.notifier.miner_stopped(self.manager)
             self.manager.close()
         if task is not None:
             try:
@@ -55,7 +59,7 @@ async def stop(self) -> bool:
 
     async def close(self) -> None:
         if self.running:
-            await self.stop()
+            await self.stop(notify=False)
 
     async def _run(self) -> None:
         success, instance_lock = lock_file(LOCK_PATH)
@@ -83,7 +87,10 @@ async def _run(self) -> None:
                 logging.getLogger("TwitchDrops.gql").setLevel(settings.debug_gql)
                 logging.getLogger("TwitchDrops.websocket").setLevel(settings.debug_ws)
                 self._logging_configured = True
-            client = Twitch(settings, gui_factory=WebManager)
+            client = Twitch(
+                settings,
+                gui_factory=lambda twitch: WebManager(twitch, self.notifier),
+            )
             self._client = client
             self.manager = client.gui
             try:
@@ -94,13 +101,19 @@ async def _run(self) -> None:
         except CaptchaRequired:
             self.last_error = _("error", "captcha")
             client.print(self.last_error)
+            if self.notifier is not None:
+                self.notifier.operational("Twitch verification required", self.last_error)
         except asyncio.CancelledError:
             raise
         except Exception:
             self.last_error = traceback.format_exc()
             if self._client is not None:
                 self._client.print("Fatal error encountered:")
                 self._client.print(self.last_error)
+            if self.notifier is not None:
+                self.notifier.operational(
+                    "Miner stopped unexpectedly", self.last_error.splitlines()[-1]
+                )
         finally:
             if self._client is not None:
                 await self._client.shutdown()
@@ -120,12 +133,14 @@ def snapshot(self) -> dict[str, Any]:
             "campaigns": [],
             "websockets": [],
             "settings": {},
+            "notifications": self.notifier.snapshot() if self.notifier is not None else {},
             "selected_channel_id": None,
             "logs": [],
         }
         if not self.running:
             state["login"]["activation_url"] = ""
             state["login"]["user_code"] = ""
+        state["notifications"] = self.notifier.snapshot() if self.notifier is not None else {}
         return {
             **state,
             "miner": {