diff --git a/.env.example b/.env.example index cab8fd5..c482a42 100644 --- a/.env.example +++ b/.env.example @@ -32,9 +32,19 @@ DOCS_CACHE_DIR=vector_db/docs_cache DOCS_HTTP_TIMEOUT=10 # Slack notifications for new support threads (optional). -# Leave SLACK_WEBHOOK_URL empty to disable them entirely. +# With neither SLACK_WEBHOOK_URL nor SLACK_BOT_TOKEN set they are disabled. +# SLACK_WEBHOOK_URL must start with https://hooks.slack.com/services/. # SUPPORT_CHANNEL_IDS: comma-separated Discord channel IDs (text or forum) whose # new threads are announced in Slack. SLACK_WEBHOOK_URL= SUPPORT_CHANNEL_IDS= SLACK_HTTP_TIMEOUT=10 + +# Slack Web API (preferred over the webhook). With a bot token the bot also +# mirrors every human message of a support thread as a reply in the Slack +# thread, which a webhook cannot do: it never returns the message ts. +# Needs scope chat:write and the app invited to the channel (/invite @app). +# SLACK_BOT_TOKEN must be a bot token (xoxb-...) and needs SLACK_CHANNEL_ID; +# otherwise token mode is disabled with one error at startup. +SLACK_BOT_TOKEN= +SLACK_CHANNEL_ID= diff --git a/AGENTS.md b/AGENTS.md index 649fa8f..eac011e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,8 +43,10 @@ ALLOWED_CHANNELS= # Channel IDs donde el bot responde (múltiple STATUS_CHANNEL_ID= # Canal donde se publica el reporte diario de métricas METRICS_SEND_HOUR=9 # Hora UTC (0-23) del reporte diario SLACK_WEBHOOK_URL= # Incoming Webhook de Slack (https://hooks.slack.com/services/...); vacío o cualquier otra URL = notificaciones deshabilitadas +SLACK_BOT_TOKEN= # Bot token (xoxb-...) con scope chat:write; habilita el espejo de hilos. Otro prefijo = modo token deshabilitado +SLACK_CHANNEL_ID= # Canal destino cuando se usa SLACK_BOT_TOKEN; obligatorio con él SUPPORT_CHANNEL_IDS= # Channel IDs cuyos hilos nuevos se avisan en Slack (coma-separados) -SLACK_HTTP_TIMEOUT=10 # Timeout (s) del POST al webhook +SLACK_HTTP_TIMEOUT=10 # Timeout (s) de las peticiones a Slack ``` **Comportamiento**: El bot SOLO responde cuando lo mencionan (`@NaN Builders`). No responde automáticamente en canales de soporte. @@ -54,13 +56,36 @@ SLACK_HTTP_TIMEOUT=10 # Timeout (s) del POST al webhook ### Notificaciones a Slack Cuando se crea un hilo en un canal listado en `SUPPORT_CHANNEL_IDS` (ej. `#support`), el bot -publica un mensaje en Slack vía Incoming Webhook con el título del hilo, el autor, el canal +publica un mensaje en Slack (vía Web API con bot token o, si no hay token, vía Incoming Webhook) con el título del hilo, el autor, el canal y un preview del mensaje inicial. Funciona con canales de texto y con canales de foro. -Requiere `SLACK_WEBHOOK_URL`: en Slack, crear una app → **Incoming Webhooks** → activar → +Modo webhook (`SLACK_WEBHOOK_URL`): en Slack, crear una app → **Incoming Webhooks** → activar → **Add New Webhook to Workspace** y elegir el canal de destino. El canal se fija ahí, no en el `.env`. -Sin `SLACK_WEBHOOK_URL` o sin `SUPPORT_CHANNEL_IDS` la feature queda inactiva y el bot funciona igual. +Sin Slack configurado o sin `SUPPORT_CHANNEL_IDS` la feature queda inactiva y el bot funciona igual. + +**Dos modos, y el bot prefiere el segundo:** + +| | Webhook (`SLACK_WEBHOOK_URL`) | Bot token (`SLACK_BOT_TOKEN` + `SLACK_CHANNEL_ID`) | +|---|---|---| +| Avisa de hilos nuevos | Sí | Sí | +| Espeja los mensajes del hilo | **No** | Sí | +| Configuración | Una URL | App con scope `chat:write` e invitada al canal | + +El webhook **no puede** espejar hilos: su respuesta no incluye el `ts` del mensaje, +y sin ese `ts` no hay forma de colgar respuestas de él. La Web API sí lo devuelve. + +Modo token: app de Slack con scope `chat:write`, instalada en el workspace e invitada al canal +(`/invite @app`). Si falta `SLACK_CHANNEL_ID`, falta el token o el token no empieza por `xoxb-`, el +modo token se deshabilita con un único error al arrancar y el bot usa el webhook si lo hay. + +El espejo se ejecuta en segundo plano (no retrasa la respuesta del bot en Discord), usa +`clean_content` (las menciones llegan como `@nombre`) y responde en el canal de Slack guardado +con el hilo. Reintenta `ratelimited`, `internal_error`, `service_unavailable`, errores de red y +respuestas no JSON; cualquier otro error de Slack se descarta sin reintentar. + +El mapeo `hilo de Discord -> ts de Slack` se guarda en `vector_db/slack_threads.db` +(dentro del volumen, así sobrevive a los despliegues) y se purga a los 30 días. ### Intents de Discord diff --git a/README.md b/README.md index 8cfce29..cead9b1 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,8 @@ Community Discord bot for [nan.builders](https://nan.builders). It answers membe - Slash commands (Discord interactions): `/metrics`, `/my-metrics`. - Daily token usage report posted to `STATUS_CHANNEL_ID` at `METRICS_SEND_HOUR` (UTC), pulled from the LiteLLM proxy. - HTTP health endpoint on port `9101` (`GET /health`) consumed by the Docker `HEALTHCHECK`. -- Slack notification on every new thread opened in the channels listed in `SUPPORT_CHANNEL_IDS` (e.g. `#support`), posted through an Incoming Webhook. +- Slack notification on every new thread opened in the channels listed in `SUPPORT_CHANNEL_IDS` (e.g. `#support`), posted through the Slack Web API (`chat.postMessage` with a bot token) or, as a fallback, an Incoming Webhook. +- Mirroring of support threads into Slack: with a bot token (not with a webhook), every human message posted in a tracked thread is echoed as a reply under the Slack announcement. - Doc-hash optimization: unchanged markdown files are skipped on startup, so embeddings are only recomputed when content actually changes. ## Tech stack @@ -35,6 +36,8 @@ Metrics live in `bot/metrics.py`. They hit the LiteLLM proxy `/spend/logs/ui` en Slack notifications live in `bot/slack.py`. `NanBot.on_thread_create` fires on Discord's `THREAD_CREATE` gateway event (only for newly created threads, so re-joins do not re-notify), filters by `SUPPORT_CHANNEL_IDS`, resolves the opening message — `Thread.starter_message` when cached, otherwise `fetch_message(thread.id)` with one retry for the forum race where the event outruns the message — and POSTs a Block Kit payload to `SLACK_WEBHOOK_URL`. The `SlackNotifier` only accepts `https://hooks.slack.com/services/` URLs (anything else disables it with one error at startup), counts a delivery only when Slack answers with the body `ok` (a 2xx with any other body is not delivered), retries 5xx and 429 with a 0.5/1/2 s backoff, never retries other 4xx, and escapes `&`, `<`, `>` in every user-controlled field so thread titles cannot forge Slack links. With `SLACK_WEBHOOK_URL` empty the notifier is a no-op. +When `SLACK_BOT_TOKEN` (an `xoxb-` bot token) and `SLACK_CHANNEL_ID` are both set, `SlackApiClient` takes over and posts through `chat.postMessage` instead. That is what makes mirroring possible: the Web API returns the message `ts`, which an Incoming Webhook never does, and a threaded reply needs it as `thread_ts`. The `ts` is stored in `vector_db/slack_threads.db` by `bot/thread_map.py` — inside the Docker volume, so a redeploy does not orphan live threads — and purged after 30 days. `NanBot._mirror_to_slack` then echoes each human message of a tracked thread (`clean_content`, so mentions read as `@name`), skipping bots and the forum opening message (already the body of the announcement). It runs as a background task, so a slow Slack never delays the bot's own answer, and replies go to the Slack channel stored with the thread, not the current `SLACK_CHANNEL_ID`. Posts are serialised behind a lock with a ~1 s spacing to respect Slack's per-channel rate limit. The Web API reports errors inside an HTTP 200 body, so `SlackApiClient` reads the `ok`/`error` fields: it retries `ratelimited`, `internal_error`, `service_unavailable`, network errors and non-JSON bodies (e.g. an HTML 502) with the same 0.5/1/2 s backoff, and gives up at once on any other error (`not_in_channel`, `invalid_auth`…). A half-configured token mode (token without channel, channel without token, or a token that is not `xoxb-`) disables it with one error at startup, and the bot falls back to the webhook if one is set. + ## Project structure ``` @@ -47,7 +50,8 @@ discord-bot/ │ ├── knowledge.py # SimpleVectorStore, chunking, doc loader │ ├── llm.py # LLMClient, CircuitBreaker, RAG prompt │ ├── metrics.py # LiteLLM spend log aggregation and reports -│ ├── slack.py # Slack Incoming Webhook notifier and payload builder +│ ├── slack.py # Slack notifier (webhook + Web API) and payload builders +│ ├── thread_map.py # Discord thread -> Slack ts mapping (SQLite) │ └── docs/ │ └── knowledge/ # Embedded markdown corpus │ ├── intro.md @@ -72,7 +76,7 @@ discord-bot/ - A Discord application with a bot user, a token, and the following privileged intents enabled in the [Discord Developer Portal](https://discord.com/developers/applications): **MESSAGE CONTENT INTENT** and **SERVER MEMBERS INTENT**. Without them the bot fails to connect with `PrivilegedIntentsRequired`. - The bot invited to your guild with permissions to read messages, send messages, embed links, and use slash commands. - A LiteLLM API key. The bot defaults to `https://api.nan.builders/v1`; override with `LITELLM_BASE_URL` if you run your own gateway. -- For the Slack support-thread notifications: a Slack app with **Incoming Webhooks** enabled and a webhook added to the destination channel, plus the Discord bot having *View Channel* and *Read Message History* on the support channel so it can read the thread's opening message for the preview. +- For the Slack support-thread notifications: either a Slack app with the `chat:write` bot scope, installed to the workspace and invited to the destination channel (`/invite @app`), which also enables thread mirroring; or a Slack app with **Incoming Webhooks** enabled and a webhook added to the destination channel (announcements only). In both cases, plus the Discord bot having *View Channel* and *Read Message History* on the support channel so it can read the thread's opening message for the preview. - For the metrics features: network reachability to the LiteLLM proxy URL (defaults to `http://localhost:4000`, i.e. the bot is expected to run on the same host) and an admin key with read access to `/spend/logs/ui`. ### Local setup (without Docker) @@ -130,7 +134,9 @@ Auto-response is triggered when the bot is **mentioned** inside a channel listed | `METRICS_SEND_HOUR` | no | `9` | UTC hour (0–23) at which the daily metrics report is posted. | | `SUPPORT_CHANNEL_IDS` | no | `""` (disables notifications) | Comma-separated Discord channel IDs (text or forum) whose new threads are announced in Slack. | | `SLACK_WEBHOOK_URL` | no | `""` (disables notifications) | Slack Incoming Webhook URL. The destination channel is fixed in the Slack app, not here. Must start with `https://hooks.slack.com/services/`; any other value disables notifications with one error at startup. | -| `SLACK_HTTP_TIMEOUT` | no | `10` | Per-request HTTP timeout (seconds) for the webhook POST. | +| `SLACK_BOT_TOKEN` | no | `""` (webhook mode) | Slack bot token (`xoxb-…`) with `chat:write`. Enables thread mirroring; takes precedence over the webhook. Any other prefix, or a missing `SLACK_CHANNEL_ID`, disables token mode with one error at startup. | +| `SLACK_CHANNEL_ID` | no | `""` | Destination Slack channel ID. Required with `SLACK_BOT_TOKEN`. | +| `SLACK_HTTP_TIMEOUT` | no | `10` | Per-request HTTP timeout (seconds) for Slack requests (webhook POST and `chat.postMessage`). | | `DOCS_USE_REMOTE` | no | `local` | Source for docs: `local` (`bot/docs/knowledge/`), `remote` (web docs API), or `shadow` (local + warn on remote drift). | | `DOCS_BASE_URL` | no | `https://nan.builders` | Base URL of the web that serves `/api/docs/manifest.json` and `/api/docs/{slug}.md`. | | `DOCS_REFRESH_INTERVAL` | no | `900` | Seconds between docs syncs when `DOCS_USE_REMOTE` is not `local`. Aligned with the web `Cache-Control`. | diff --git a/bot/base.py b/bot/base.py index 26d4775..6272571 100644 --- a/bot/base.py +++ b/bot/base.py @@ -7,6 +7,7 @@ from collections import defaultdict from datetime import UTC from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path from threading import Thread import discord @@ -17,7 +18,8 @@ from bot.knowledge import SimpleVectorStore, load_documentation_from_remote from bot.llm import LLMClient from bot.metrics import send_metrics_report, send_user_metrics_report -from bot.slack import SlackNotifier, SupportThreadEvent +from bot.slack import SlackApiClient, SlackNotifier, SupportThreadEvent +from bot.thread_map import SlackThreadMap # Rate limiting: max 3 mentions per user per 60-second window _RATE_LIMIT = 3 @@ -69,7 +71,15 @@ def __init__(self) -> None: ) self.llm = LLMClient() + # The Web API is preferred: only it returns the message ts that thread + # replies need. The webhook stays as a fallback so a deployment without + # a bot token keeps announcing threads, just without mirroring. self.slack = SlackNotifier() + self.slack_api = SlackApiClient() + # Same directory main.py gives the vector store: it is the Docker + # volume, so the mapping survives redeploys. (config.DB_DIR points at + # bot/vector_db, which is not mounted, and is unused.) + self.thread_map = SlackThreadMap(Path("vector_db")) self.store: SimpleVectorStore | None = None self._initialized = False self._ready = False @@ -82,6 +92,9 @@ def __init__(self) -> None: self._docs_last_sync_ok: bool = False self._docs_refresh_task: asyncio.Task[None] | None = None self._docs_refresh_lock = asyncio.Lock() + # Strong references to fire-and-forget tasks (the Slack mirror), so the + # event loop's weak references do not let them be garbage-collected. + self._bg_tasks: set[asyncio.Task[None]] = set() def _start_health_server(self) -> None: """Start a lightweight HTTP health check server in a background thread.""" @@ -159,6 +172,11 @@ async def on_ready(self) -> None: ) await self.start_daily_metrics() + try: + self.thread_map.purge_older_than() + except Exception as e: + logger.warning("Could not purge stale Slack thread mappings: %s", type(e).__name__) + if self._docs_refresh_task is None or self._docs_refresh_task.done(): self._docs_refresh_task = asyncio.create_task(self._schedule_docs_refresh()) else: @@ -282,8 +300,8 @@ async def on_thread_create(self, thread: discord.Thread) -> None: if not support_channels or thread.parent_id not in support_channels: return - if not self.slack.enabled: - logger.warning("New thread in support channel but SLACK_WEBHOOK_URL is not configured") + if not (self.slack_api.enabled or self.slack.enabled): + logger.warning("New thread in a support channel but Slack is not configured") return preview = await self._fetch_starter_text(thread) @@ -302,7 +320,12 @@ async def on_thread_create(self, thread: discord.Thread) -> None: ) try: - sent = await asyncio.wait_for(self.slack.notify_support_thread(event), timeout=30.0) + if self.slack_api.enabled: + ts = await asyncio.wait_for(self.slack_api.notify_support_thread(event), timeout=30.0) + sent = ts is not None + else: + ts = None + sent = await asyncio.wait_for(self.slack.notify_support_thread(event), timeout=30.0) except TimeoutError: logger.error("Slack notification timed out for thread %s", thread.id) return @@ -310,14 +333,84 @@ async def on_thread_create(self, thread: discord.Thread) -> None: logger.error("Slack notification failed for thread %s: %s", thread.id, type(e).__name__) return + if ts: + # Without this the thread is announced but every later message has + # nothing to hang off. Slack already has the message, so a failure + # here is a storage problem, not a notification one. + try: + self.thread_map.remember(thread.id, ts, self.slack_api.channel) + except Exception as e: + logger.error( + "Thread %s was announced in Slack but could not store the thread mapping: %s", + thread.id, + type(e).__name__, + ) + if sent: logger.info("Notified Slack about thread %s in #%s", thread.id, channel_name) + async def _mirror_to_slack(self, message: discord.Message) -> None: + """Mirror a message posted in a tracked support thread into Slack.""" + # The bot's own answers are long LLM replies; Slack mirrors the human + # conversation only. Other bots are skipped for the same reason. + if message.author.bot: + return + + channel = message.channel + if not isinstance(channel, discord.Thread): + return + if channel.parent_id not in settings.support_channel_id_set: + return + + # A forum post's opening message arrives here too, but it is already the + # body of the announcement, so mirroring it would duplicate it. + if message.id == channel.id: + return + + mapping = self.thread_map.lookup(channel.id) + if mapping is None: + # Threads opened before mirroring existed, or announced while Slack + # was down, have no parent to reply to. + logger.debug("No Slack thread mapped for Discord thread %s", channel.id) + return + + thread_ts, slack_channel = mapping + # clean_content turns <@123…> mentions into readable @names; Slack + # cannot resolve Discord IDs. + content = message.clean_content or "" + if message.attachments: + names = ", ".join(a.filename for a in message.attachments[:5]) + content = f"{content}\n[attachments: {names}]".strip() + if not content: + return + + try: + await asyncio.wait_for( + self.slack_api.reply_in_thread( + thread_ts, + _sanitize_username(message.author.display_name), + content, + channel=slack_channel, + ), + timeout=30.0, + ) + except TimeoutError: + logger.error("Slack mirror timed out for message %s", message.id) + except Exception as e: + logger.error("Slack mirror failed for message %s: %s", message.id, type(e).__name__) + async def on_message(self, message: discord.Message) -> None: """Process incoming messages for auto-responses.""" if message.author == self.user: return + if self.slack_api.enabled: + # In the background: a slow Slack (30 s timeout plus retries) must + # not delay the bot's own answer in Discord. + task = asyncio.create_task(self._mirror_to_slack(message)) + self._bg_tasks.add(task) + task.add_done_callback(self._bg_tasks.discard) + # Delegate to slash commands before processing auto-responses await self.process_commands(message) diff --git a/bot/config.py b/bot/config.py index 51e9af7..dfbce28 100644 --- a/bot/config.py +++ b/bot/config.py @@ -43,6 +43,8 @@ class Settings(BaseSettings): docs_http_timeout: int = 10 slack_webhook_url: str = "" + slack_bot_token: str = "" + slack_channel_id: str = "" slack_http_timeout: int = 10 support_channel_ids: str = "" diff --git a/bot/slack.py b/bot/slack.py index ce104d9..656b8c7 100644 --- a/bot/slack.py +++ b/bot/slack.py @@ -1,4 +1,4 @@ -"""Slack notifications via Incoming Webhook.""" +"""Slack notifications via Incoming Webhook or the Web API (chat.postMessage).""" from __future__ import annotations @@ -17,6 +17,14 @@ _WEBHOOK_PREFIX = "https://hooks.slack.com/services/" _MAX_TITLE_LEN = 200 _MAX_PREVIEW_LEN = 600 +# Slack rejects a section block whose text exceeds 3000 characters. The limit +# applies to the escaped text, and escaping can grow it (one "<" becomes "<"), +# so the mirrored body is truncated after escaping, with headroom to spare. +_MAX_SECTION_LEN = 3000 +_MAX_REPLY_LEN = 2900 +# A bot token, not a user (xoxp-) or app-level (xapp-) one: only a bot token +# posts as the app with chat:write. +_BOT_TOKEN_PREFIX = "xoxb-" # Slack mrkdwn requires these three characters to be HTML-escaped so user text # cannot forge links or entities inside a block. @@ -37,6 +45,22 @@ def _truncate(text: str, limit: int) -> str: return text[: limit - 1].rstrip() + "…" +def _truncate_escaped(text: str, limit: int) -> str: + """Truncate already-escaped mrkdwn without cutting an entity in half. + + After escape_mrkdwn every "&" starts an entity ("&", "<", ">"), so + a "&" with no ";" after it in the kept part is a broken entity to drop. + """ + text = text.strip() + if len(text) <= limit: + return text + cut = text[: limit - 1] + amp = cut.rfind("&") + if amp != -1 and ";" not in cut[amp:]: + cut = cut[:amp] + return cut.rstrip() + "…" + + @dataclass class SupportThreadEvent: """The data needed to announce a new Discord support thread in Slack.""" @@ -87,6 +111,25 @@ def build_support_thread_payload(event: SupportThreadEvent) -> dict: } +def build_reply_payload(author: str, text: str) -> dict: + """Build the Slack payload mirroring one Discord message into a thread.""" + safe_author = escape_mrkdwn(_truncate(author, _MAX_TITLE_LEN)) or "unknown" + header = f"*{safe_author}*\n" + # Escaped once and used in both places: the fallback text is what Slack + # shows in notifications, so it must not carry or forged links. + body = _truncate_escaped(escape_mrkdwn(text), min(_MAX_REPLY_LEN, _MAX_SECTION_LEN - len(header))) + + return { + "text": f"{safe_author}: {body}", + "blocks": [ + { + "type": "section", + "text": {"type": "mrkdwn", "text": f"{header}{body}"}, + } + ], + } + + class SlackNotifier: """Posts messages to a Slack Incoming Webhook. @@ -174,3 +217,135 @@ async def post(self, payload: dict) -> bool: async def notify_support_thread(self, event: SupportThreadEvent) -> bool: """Announce a new Discord support thread in Slack.""" return await self.post(build_support_thread_payload(event)) + + +class SlackApiClient: + """Posts through chat.postMessage with a bot token. + + Unlike an Incoming Webhook, the Web API returns the message ``ts``, which is + what a threaded reply needs as ``thread_ts``. Slack allows roughly one + message per second per channel, so posts are serialised behind a lock. + """ + + _URL = "https://slack.com/api/chat.postMessage" + _MIN_INTERVAL = 1.05 + + def __init__( + self, + bot_token: str | None = None, + channel: str | None = None, + timeout: float | None = None, + ) -> None: + self._token = (bot_token if bot_token is not None else settings.slack_bot_token).strip() + self._channel = (channel if channel is not None else settings.slack_channel_id).strip() + self._timeout = timeout if timeout is not None else float(settings.slack_http_timeout) + self._client: httpx.AsyncClient | None = None + self._lock = asyncio.Lock() + self._last_post = float("-inf") + # Validated once, at startup, like the webhook URL: a half-configured + # token mode would otherwise fall back to the webhook without a word. + self._enabled = self._validate() + + def _validate(self) -> bool: + if not self._token and not self._channel: + return False + if not self._token: + logger.error("SLACK_CHANNEL_ID is set but SLACK_BOT_TOKEN is empty; thread mirroring is disabled") + return False + if not self._token.startswith(_BOT_TOKEN_PREFIX): + logger.error( + "SLACK_BOT_TOKEN is not a bot token (must start with %s); thread mirroring is disabled", + _BOT_TOKEN_PREFIX, + ) + return False + if not self._channel: + logger.error("SLACK_BOT_TOKEN is set but SLACK_CHANNEL_ID is empty; thread mirroring is disabled") + return False + return True + + @property + def enabled(self) -> bool: + return self._enabled + + @property + def channel(self) -> str: + return self._channel + + def _get_client(self) -> httpx.AsyncClient: + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient(timeout=self._timeout) + return self._client + + async def close(self) -> None: + if self._client is not None and not self._client.is_closed: + await self._client.aclose() + self._client = None + + async def _throttle(self) -> None: + """Space out posts to stay under Slack's per-channel rate limit.""" + loop = asyncio.get_running_loop() + elapsed = loop.time() - self._last_post + if elapsed < self._MIN_INTERVAL: + await asyncio.sleep(self._MIN_INTERVAL - elapsed) + self._last_post = loop.time() + + async def post(self, payload: dict, thread_ts: str | None = None, channel: str | None = None) -> str | None: + """Post a message. Returns its ``ts``, or None when it was not delivered. + + ``channel`` overrides the configured one: a reply must go to the channel + that holds its parent, even if SLACK_CHANNEL_ID changed since. + """ + if not self.enabled: + logger.debug("Slack bot token or channel not configured, skipping") + return None + + body = {**payload, "channel": channel or self._channel} + if thread_ts: + body["thread_ts"] = thread_ts + + headers = {"Authorization": f"Bearer {self._token}"} + client = self._get_client() + last_error: str | None = None + + for attempt, backoff in enumerate((*_BACKOFF_SECONDS, None)): + async with self._lock: + await self._throttle() + try: + resp = await client.post(self._URL, json=body, headers=headers) + except httpx.HTTPError as e: + last_error = type(e).__name__ + resp = None + + if resp is not None: + # The Web API reports failures inside a 200 body, so the status + # code alone never tells you whether the message landed. + try: + data = resp.json() + except ValueError: + last_error = f"HTTP {resp.status_code} with a non-JSON body" + data = {} + else: + if data.get("ok"): + return data.get("ts") + error = data.get("error", "unknown_error") + if error not in ("ratelimited", "internal_error", "service_unavailable"): + logger.error("Slack rejected the message: %s", error) + return None + last_error = error + + if backoff is None: + break + logger.warning("Slack API attempt %d failed (%s), retrying", attempt + 1, last_error) + await asyncio.sleep(backoff) + + logger.error("Slack API failed after %d attempts: %s", len(_BACKOFF_SECONDS) + 1, last_error) + return None + + async def notify_support_thread(self, event: SupportThreadEvent) -> str | None: + """Announce a new Discord support thread. Returns the Slack ``ts``.""" + return await self.post(build_support_thread_payload(event)) + + async def reply_in_thread(self, thread_ts: str, author: str, text: str, channel: str | None = None) -> bool: + """Mirror one Discord message as a reply under the announcement.""" + ts = await self.post(build_reply_payload(author, text), thread_ts=thread_ts, channel=channel) + return ts is not None diff --git a/bot/thread_map.py b/bot/thread_map.py new file mode 100644 index 0000000..152f4b8 --- /dev/null +++ b/bot/thread_map.py @@ -0,0 +1,72 @@ +"""Persistent mapping from a Discord thread to its Slack message. + +Slack replies are threaded by passing the parent message's ``ts``. That value +only exists after the parent has been posted, so it has to outlive the process: +without it, a restart orphans every live support thread. +""" + +from __future__ import annotations + +import sqlite3 +import time +from pathlib import Path + +from bot.config import logger + +# Discord archives support threads long before this, and a Slack thread that old +# is not worth replying into. Keeps the table from growing without bound. +_RETENTION_SECONDS = 30 * 24 * 3600 + + +class SlackThreadMap: + """Maps Discord thread IDs to the Slack message ts that announced them.""" + + def __init__(self, db_dir: Path) -> None: + db_dir.mkdir(parents=True, exist_ok=True) + self._db_path = db_dir / "slack_threads.db" + self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._init_schema() + + def _init_schema(self) -> None: + self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS slack_threads ( + discord_thread_id INTEGER PRIMARY KEY, + slack_ts TEXT NOT NULL, + slack_channel TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + """) + self._conn.commit() + + def remember(self, discord_thread_id: int, slack_ts: str, slack_channel: str) -> None: + """Record the Slack message that announced a Discord thread.""" + self._conn.execute( + "INSERT OR REPLACE INTO slack_threads " + "(discord_thread_id, slack_ts, slack_channel, created_at) VALUES (?, ?, ?, ?)", + (discord_thread_id, slack_ts, slack_channel, int(time.time())), + ) + self._conn.commit() + + def lookup(self, discord_thread_id: int) -> tuple[str, str] | None: + """Return (slack_ts, slack_channel) for a thread, or None if unknown.""" + row = self._conn.execute( + "SELECT slack_ts, slack_channel FROM slack_threads WHERE discord_thread_id = ?", + (discord_thread_id,), + ).fetchone() + if row is None: + return None + return row["slack_ts"], row["slack_channel"] + + def purge_older_than(self, seconds: int = _RETENTION_SECONDS) -> int: + """Drop mappings older than `seconds`. Returns how many were removed.""" + cutoff = int(time.time()) - seconds + cursor = self._conn.execute("DELETE FROM slack_threads WHERE created_at < ?", (cutoff,)) + self._conn.commit() + removed = cursor.rowcount + if removed: + logger.info("Purged %d stale Slack thread mappings", removed) + return removed + + def close(self) -> None: + self._conn.close() diff --git a/main.py b/main.py index f2d0405..d20a5c2 100644 --- a/main.py +++ b/main.py @@ -134,12 +134,22 @@ async def main() -> None: await bot.llm._embed_client.close() except Exception: pass - # Close Slack client + # Close Slack clients and the thread mapping if bot.slack: try: await bot.slack.close() except Exception: pass + if bot.slack_api: + try: + await bot.slack_api.close() + except Exception: + pass + if bot.thread_map: + try: + bot.thread_map.close() + except Exception: + pass # Stop health check server if hasattr(bot, "_health_server") and bot._health_server: try: diff --git a/tests/test_on_thread_create.py b/tests/test_on_thread_create.py index 4ce2b05..358e00a 100644 --- a/tests/test_on_thread_create.py +++ b/tests/test_on_thread_create.py @@ -2,6 +2,9 @@ from __future__ import annotations +import asyncio +import logging +import sqlite3 from dataclasses import dataclass, field import discord @@ -72,6 +75,12 @@ async def notify_support_thread(self, event) -> bool: return self._result +@pytest.fixture(autouse=True) +def isolated_cwd(tmp_path, monkeypatch): + """NanBot opens the thread-map DB relative to the cwd; keep it out of the repo.""" + monkeypatch.chdir(tmp_path) + + @pytest.fixture(autouse=True) def no_sleep(monkeypatch): """Keep the starter-message retry backoff from slowing the suite down.""" @@ -87,7 +96,8 @@ def bot(monkeypatch): monkeypatch.setattr(settings, "support_channel_ids", str(SUPPORT_CHANNEL_ID)) instance = NanBot() instance.slack = RecordingSlack() - return instance + yield instance + instance.thread_map.close() async def test_notifies_slack_for_a_thread_in_a_support_channel(bot): @@ -117,6 +127,7 @@ async def test_ignores_every_thread_when_no_support_channel_is_configured(monkey await instance.on_thread_create(FakeThread()) assert instance.slack.events == [] + instance.thread_map.close() async def test_does_nothing_when_slack_is_not_configured(bot): @@ -184,3 +195,293 @@ async def notify_support_thread(self, event) -> bool: bot.slack = ExplodingSlack() await bot.on_thread_create(FakeThread(starter_message=FakeMessage("hola"))) + + +# --- mirroring later messages into the Slack thread ----------------------- + + +@dataclass +class FakeAuthor: + display_name: str = "crstian" + bot: bool = False + + +@dataclass +class FakeAttachment: + filename: str + + +@dataclass +class FakeDiscordMessage: + id: int = 777 + content: str = "sigue sin arrancar" + author: FakeAuthor = field(default_factory=FakeAuthor) + channel: object = None + attachments: list = field(default_factory=list) + # discord.Message.clean_content resolves <@123> into @name; the fake mirrors + # content unless a test sets it. + clean_content: str | None = None + + def __post_init__(self) -> None: + if self.clean_content is None: + self.clean_content = self.content + + +class RecordingApi: + def __init__(self, enabled: bool = True, ts: str | None = "1610144875.000600") -> None: + self.enabled = enabled + self.channel = "C0123" + self._ts = ts + self.replies: list[tuple[str, str, str]] = [] + self.reply_channels: list[str | None] = [] + self.announced: list[object] = [] + + async def notify_support_thread(self, event): + self.announced.append(event) + return self._ts + + async def reply_in_thread(self, thread_ts: str, author: str, text: str, channel: str | None = None) -> bool: + self.replies.append((thread_ts, author, text)) + self.reply_channels.append(channel) + return True + + +@pytest.fixture +def mirroring_bot(bot): + bot.slack_api = RecordingApi() + return bot + + +def _thread_channel(bot, thread_id: int = 555, parent_id: int = SUPPORT_CHANNEL_ID): + """A discord.Thread stand-in that passes the isinstance check.""" + channel = discord.Thread.__new__(discord.Thread) + object.__setattr__(channel, "id", thread_id) + object.__setattr__(channel, "parent_id", parent_id) + return channel + + +async def test_announcing_a_thread_stores_the_slack_ts(mirroring_bot): + thread = FakeThread(starter_message=FakeMessage("hola")) + + await mirroring_bot.on_thread_create(thread) + + assert mirroring_bot.thread_map.lookup(thread.id) == ("1610144875.000600", "C0123") + + +async def test_a_message_in_a_tracked_thread_is_mirrored(mirroring_bot): + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(channel=_thread_channel(mirroring_bot)) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies == [("1610144875.000600", "crstian", "sigue sin arrancar")] + + +async def test_the_forum_opening_message_is_not_mirrored_twice(mirroring_bot): + """It is already the body of the announcement.""" + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(id=555, channel=_thread_channel(mirroring_bot, thread_id=555)) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies == [] + + +async def test_a_thread_with_no_mapping_is_skipped(mirroring_bot): + msg = FakeDiscordMessage(channel=_thread_channel(mirroring_bot)) + await mirroring_bot._mirror_to_slack(msg) + assert mirroring_bot.slack_api.replies == [] + + +async def test_messages_outside_support_channels_are_skipped(mirroring_bot): + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(channel=_thread_channel(mirroring_bot, parent_id=OTHER_CHANNEL_ID)) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies == [] + + +async def test_attachments_are_named_in_the_mirrored_message(mirroring_bot): + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage( + content="mira esto", + channel=_thread_channel(mirroring_bot), + attachments=[FakeAttachment("error.png")], + ) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies[0][2] == "mira esto\n[attachments: error.png]" + + +async def test_an_empty_message_is_not_mirrored(mirroring_bot): + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(content="", channel=_thread_channel(mirroring_bot)) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies == [] + + +async def test_a_mirror_failure_does_not_propagate(mirroring_bot): + class ExplodingApi(RecordingApi): + async def reply_in_thread(self, thread_ts, author, text): + raise RuntimeError("slack down") + + mirroring_bot.slack_api = ExplodingApi() + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + + await mirroring_bot._mirror_to_slack(FakeDiscordMessage(channel=_thread_channel(mirroring_bot))) + + +async def test_bot_messages_are_not_mirrored(mirroring_bot): + """Slack mirrors the human conversation, not the bot's LLM answers.""" + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage( + author=FakeAuthor(display_name="NaN Builders", bot=True), + channel=_thread_channel(mirroring_bot), + ) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies == [] + + +async def test_mentions_reach_slack_as_names_not_raw_ids(mirroring_bot): + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage( + content="<@123456789012345678> mira esto", + clean_content="@crstian mira esto", + channel=_thread_channel(mirroring_bot), + ) + + await mirroring_bot._mirror_to_slack(msg) + + assert mirroring_bot.slack_api.replies[0][2] == "@crstian mira esto" + + +async def test_a_reply_goes_to_the_channel_stored_with_the_thread(mirroring_bot): + """SLACK_CHANNEL_ID may have changed since the announcement; the parent did not move.""" + mirroring_bot.thread_map.remember(555, "1610144875.000600", "C_OLD") + + await mirroring_bot._mirror_to_slack(FakeDiscordMessage(channel=_thread_channel(mirroring_bot))) + + assert mirroring_bot.slack_api.reply_channels == ["C_OLD"] + + +async def test_a_failed_announcement_stores_no_mapping(mirroring_bot, monkeypatch): + mirroring_bot.slack_api = RecordingApi(ts=None) + stored: list[tuple] = [] + monkeypatch.setattr(mirroring_bot.thread_map, "remember", lambda *args: stored.append(args)) + + await mirroring_bot.on_thread_create(FakeThread(starter_message=FakeMessage("hola"))) + + assert stored == [] + + +async def test_a_mapping_failure_is_not_reported_as_a_slack_failure(mirroring_bot, monkeypatch, caplog): + def broken(*_args): + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr(mirroring_bot.thread_map, "remember", broken) + + with caplog.at_level(logging.INFO, logger="nan-bot"): + await mirroring_bot.on_thread_create(FakeThread(starter_message=FakeMessage("hola"))) + + assert "could not store the thread mapping" in caplog.text + assert "Slack notification failed" not in caplog.text + + +# --- on_message wiring and on_ready housekeeping -------------------------- + + +class FakeUser: + id = 1 + + def mentioned_in(self, _message) -> bool: + return False + + +@pytest.fixture +def connected_bot(mirroring_bot): + """A bot that looks logged in, with commands stubbed out.""" + mirroring_bot._connection.user = FakeUser() + mirroring_bot.processed: list[object] = [] + + async def process_commands(message) -> None: + mirroring_bot.processed.append(message) + + mirroring_bot.process_commands = process_commands + return mirroring_bot + + +async def _drain(bot) -> None: + if bot._bg_tasks: + await asyncio.gather(*bot._bg_tasks) + + +async def test_on_message_mirrors_a_reply_in_a_tracked_thread(connected_bot): + connected_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(channel=_thread_channel(connected_bot)) + + await connected_bot.on_message(msg) + await _drain(connected_bot) + + assert connected_bot.slack_api.replies == [("1610144875.000600", "crstian", "sigue sin arrancar")] + assert connected_bot.processed == [msg] + + +async def test_on_message_mirrors_nothing_without_the_web_api(connected_bot): + connected_bot.slack_api = RecordingApi(enabled=False) + connected_bot.thread_map.remember(555, "1610144875.000600", "C0123") + + await connected_bot.on_message(FakeDiscordMessage(channel=_thread_channel(connected_bot))) + await _drain(connected_bot) + + assert connected_bot.slack_api.replies == [] + assert connected_bot._bg_tasks == set() + + +async def test_a_slow_slack_does_not_delay_the_bot(connected_bot): + release = asyncio.Event() + + class SlowApi(RecordingApi): + async def reply_in_thread(self, *args, **kwargs) -> bool: + await release.wait() + return await super().reply_in_thread(*args, **kwargs) + + connected_bot.slack_api = SlowApi() + connected_bot.thread_map.remember(555, "1610144875.000600", "C0123") + msg = FakeDiscordMessage(channel=_thread_channel(connected_bot)) + + # Awaiting the mirror inline would block here until Slack answered. + await asyncio.wait_for(connected_bot.on_message(msg), timeout=1.0) + assert connected_bot.processed == [msg] + assert connected_bot.slack_api.replies == [] + + release.set() + await _drain(connected_bot) + assert len(connected_bot.slack_api.replies) == 1 + assert connected_bot._bg_tasks == set() + + +async def test_on_ready_purges_stale_thread_mappings(connected_bot, monkeypatch): + async def noop(*_args, **_kwargs) -> None: + return None + + monkeypatch.setattr(connected_bot, "change_presence", noop) + monkeypatch.setattr(connected_bot, "start_daily_metrics", noop) + monkeypatch.setattr(connected_bot, "_schedule_docs_refresh", noop) + + tm = connected_bot.thread_map + tm.remember(1, "1.1", "C0123") + tm.remember(2, "2.2", "C0123") + tm._conn.execute("UPDATE slack_threads SET created_at = 0 WHERE discord_thread_id = 1") + tm._conn.commit() + + await connected_bot.on_ready() + await connected_bot._docs_refresh_task + + assert tm.lookup(1) is None + assert tm.lookup(2) == ("2.2", "C0123") diff --git a/tests/test_slack_api.py b/tests/test_slack_api.py new file mode 100644 index 0000000..c27c5c7 --- /dev/null +++ b/tests/test_slack_api.py @@ -0,0 +1,299 @@ +"""Tests for the Slack Web API client used to thread replies.""" + +from __future__ import annotations + +import asyncio +import json +import logging + +import httpx +import pytest + +from bot.slack import SlackApiClient, SupportThreadEvent, build_reply_payload + +EVENT = SupportThreadEvent( + thread_name="El bot no responde", + thread_url="https://discord.com/channels/1/2", + channel_name="support", + author="crstian", +) + + +def _client(handler, **kwargs) -> SlackApiClient: + api = SlackApiClient(bot_token="xoxb-test", channel="C0123", **kwargs) + api._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + api._MIN_INTERVAL = 0 # no throttling in tests + return api + + +async def test_disabled_without_a_token_or_channel(): + assert SlackApiClient(bot_token="", channel="C0123").enabled is False + assert SlackApiClient(bot_token="xoxb-test", channel="").enabled is False + assert SlackApiClient(bot_token="xoxb-test", channel="C0123").enabled is True + + +async def test_announcing_a_thread_returns_the_ts(): + async def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["Authorization"] == "Bearer xoxb-test" + body = json.loads(request.content) + assert body["channel"] == "C0123" + assert "thread_ts" not in body + return httpx.Response(200, json={"ok": True, "ts": "1610144875.000600"}) + + api = _client(handler) + assert await api.notify_support_thread(EVENT) == "1610144875.000600" + await api.close() + + +async def test_a_reply_carries_thread_ts(): + seen = {} + + async def handler(request: httpx.Request) -> httpx.Response: + seen.update(json.loads(request.content)) + return httpx.Response(200, json={"ok": True, "ts": "1610144999.000100"}) + + api = _client(handler) + assert await api.reply_in_thread("1610144875.000600", "crstian", "no arranca") is True + assert seen["thread_ts"] == "1610144875.000600" + assert "crstian" in json.dumps(seen) + await api.close() + + +async def test_an_error_inside_a_200_is_not_a_delivery(): + """The Web API reports failures in the body, with HTTP 200.""" + + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": False, "error": "channel_not_found"}) + + api = _client(handler) + assert await api.notify_support_thread(EVENT) is None + await api.close() + + +async def test_permanent_errors_are_not_retried(): + calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(200, json={"ok": False, "error": "not_in_channel"}) + + api = _client(handler) + assert await api.notify_support_thread(EVENT) is None + assert calls == 1 + await api.close() + + +async def test_ratelimited_is_retried(monkeypatch): + monkeypatch.setattr("bot.slack._BACKOFF_SECONDS", (0, 0, 0)) + calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls < 3: + return httpx.Response(200, json={"ok": False, "error": "ratelimited"}) + return httpx.Response(200, json={"ok": True, "ts": "1.2"}) + + api = _client(handler) + assert await api.notify_support_thread(EVENT) == "1.2" + assert calls == 3 + await api.close() + + +@pytest.mark.parametrize( + ("author", "text", "expected"), + [ + ("crstian", "hola", "*crstian*\nhola"), + ("<@here>", "a > b", "*<@here>*\na > b"), + ], +) +def test_reply_payload_escapes_user_content(author, text, expected): + payload = build_reply_payload(author, text) + assert payload["blocks"][0]["text"]["text"] == expected + # The fallback text is what Slack puts in notifications. + author_line, body = expected.split("\n", 1) + assert payload["text"] == f"{author_line.strip('*')}: {body}" + + +def test_reply_fallback_text_cannot_ping_the_channel_or_forge_links(): + payload = build_reply_payload("crstian", " look ") + assert "<" not in payload["text"] + assert ">" not in payload["text"] + assert "<!channel>" in payload["text"] + + +def test_reply_payload_truncates_long_messages(): + payload = build_reply_payload("crstian", "x" * 5000) + assert len(payload["blocks"][0]["text"]["text"]) <= 3000 + assert "…" in payload["blocks"][0]["text"]["text"] + + +@pytest.mark.parametrize("char", ["<", ">", "&"]) +def test_escaping_cannot_push_a_reply_past_the_section_limit(char): + """One "<" becomes "<": truncating before escaping lets 2000 chars grow to 8000.""" + payload = build_reply_payload("a", char * 2000) + text = payload["blocks"][0]["text"]["text"] + + assert len(text) <= 3000 + body = text.split("\n", 1)[1].removesuffix("…") + entity = {"<": "<", ">": ">", "&": "&"}[char] + # Only whole entities: nothing is left of a half-cut "<". + assert body.replace(entity, "") == "" + + +def test_a_long_author_still_fits_the_section_limit(): + payload = build_reply_payload("&" * 500, "<" * 2000) + assert len(payload["blocks"][0]["text"]["text"]) <= 3000 + + +# --- configuration -------------------------------------------------------- + + +@pytest.mark.parametrize( + ("token", "channel", "message"), + [ + ("xoxb-test", "", "SLACK_CHANNEL_ID is empty"), + ("", "C0123", "SLACK_BOT_TOKEN is empty"), + ("xoxp-user-token", "C0123", "not a bot token"), + ("https://hooks.slack.com/services/T/B/X", "C0123", "not a bot token"), + ], +) +def test_a_half_configured_token_mode_is_one_error_at_startup(token, channel, message, caplog): + with caplog.at_level(logging.ERROR, logger="nan-bot"): + api = SlackApiClient(bot_token=token, channel=channel) + + assert api.enabled is False + errors = [r for r in caplog.records if r.levelno == logging.ERROR] + assert len(errors) == 1 + assert message in errors[0].getMessage() + # The token itself must never reach the logs. + if token: + assert token not in caplog.text + + +def test_no_token_and_no_channel_is_silent(caplog): + with caplog.at_level(logging.DEBUG, logger="nan-bot"): + assert SlackApiClient(bot_token="", channel="").enabled is False + assert caplog.records == [] + + +async def test_a_reply_goes_to_the_requested_channel(): + seen = {} + + async def handler(request: httpx.Request) -> httpx.Response: + seen.update(json.loads(request.content)) + return httpx.Response(200, json={"ok": True, "ts": "2.2"}) + + api = _client(handler) + assert await api.reply_in_thread("1.1", "crstian", "hola", channel="C_OLD") is True + assert seen["channel"] == "C_OLD" + await api.close() + + +# --- retries --------------------------------------------------------------- + + +@pytest.mark.parametrize( + "failure", + [ + httpx.Response(200, json={"ok": False, "error": "internal_error"}), + httpx.Response(200, json={"ok": False, "error": "service_unavailable"}), + httpx.ConnectError("connection refused"), + httpx.Response(502, text="502 Bad Gateway"), + ], + ids=["internal_error", "service_unavailable", "connect_error", "502_html"], +) +async def test_transient_failures_are_retried(failure, monkeypatch): + monkeypatch.setattr("bot.slack._BACKOFF_SECONDS", (0, 0, 0)) + calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls == 1: + if isinstance(failure, Exception): + raise failure + return failure + return httpx.Response(200, json={"ok": True, "ts": "1.2"}) + + api = _client(handler) + assert await api.notify_support_thread(EVENT) == "1.2" + assert calls == 2 + await api.close() + + +async def test_gives_up_after_four_attempts(monkeypatch): + monkeypatch.setattr("bot.slack._BACKOFF_SECONDS", (0, 0, 0)) + calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + raise httpx.ConnectError("connection refused") + + api = _client(handler) + assert await api.notify_support_thread(EVENT) is None + assert calls == 4 + await api.close() + + +# --- throttling ------------------------------------------------------------ + + +@pytest.fixture +def recorded_sleeps(monkeypatch): + """Record asyncio.sleep calls without waiting, but still yield to the loop.""" + real_sleep = asyncio.sleep + waits: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + waits.append(seconds) + await real_sleep(0) + + monkeypatch.setattr("bot.slack.asyncio.sleep", fake_sleep) + return waits, real_sleep + + +async def test_consecutive_posts_are_spaced_by_the_channel_rate_limit(recorded_sleeps): + waits, _ = recorded_sleeps + + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True, "ts": "1.2"}) + + api = _client(handler) + api._MIN_INTERVAL = 1.05 + + await api.notify_support_thread(EVENT) + assert waits == [] # the first post goes out immediately + await api.notify_support_thread(EVENT) + + assert len(waits) == 1 + assert waits[0] == pytest.approx(1.05, abs=0.05) + await api.close() + + +async def test_concurrent_posts_are_serialised(recorded_sleeps): + waits, real_sleep = recorded_sleeps + in_flight = 0 + peak = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal in_flight, peak + in_flight += 1 + peak = max(peak, in_flight) + for _ in range(5): + await real_sleep(0) + in_flight -= 1 + return httpx.Response(200, json={"ok": True, "ts": "1.2"}) + + api = _client(handler) + api._MIN_INTERVAL = 1.05 + + results = await asyncio.gather(api.notify_support_thread(EVENT), api.notify_support_thread(EVENT)) + + assert results == ["1.2", "1.2"] + assert peak == 1 + assert len(waits) == 1 + assert waits[0] == pytest.approx(1.05, abs=0.05) + await api.close() diff --git a/tests/test_thread_map.py b/tests/test_thread_map.py new file mode 100644 index 0000000..5d62410 --- /dev/null +++ b/tests/test_thread_map.py @@ -0,0 +1,58 @@ +"""Tests for the Discord thread -> Slack message mapping.""" + +from __future__ import annotations + +import time + +from bot.thread_map import SlackThreadMap + + +def test_remembers_and_looks_up_a_mapping(tmp_path): + tm = SlackThreadMap(tmp_path) + tm.remember(555, "1610144875.000600", "C0123") + + assert tm.lookup(555) == ("1610144875.000600", "C0123") + tm.close() + + +def test_unknown_thread_returns_none(tmp_path): + tm = SlackThreadMap(tmp_path) + assert tm.lookup(999) is None + tm.close() + + +def test_mapping_survives_reopening_the_database(tmp_path): + """A redeploy restarts the process; the mapping has to outlive it.""" + tm = SlackThreadMap(tmp_path) + tm.remember(555, "1610144875.000600", "C0123") + tm.close() + + reopened = SlackThreadMap(tmp_path) + assert reopened.lookup(555) == ("1610144875.000600", "C0123") + reopened.close() + + +def test_remembering_the_same_thread_twice_overwrites(tmp_path): + tm = SlackThreadMap(tmp_path) + tm.remember(555, "111.000", "C0123") + tm.remember(555, "222.000", "C0123") + + assert tm.lookup(555) == ("222.000", "C0123") + tm.close() + + +def test_purge_drops_only_old_mappings(tmp_path): + tm = SlackThreadMap(tmp_path) + tm.remember(1, "111.000", "C0123") + tm.remember(2, "222.000", "C0123") + # Backdate one row by 40 days. + tm._conn.execute( + "UPDATE slack_threads SET created_at = ? WHERE discord_thread_id = 1", + (int(time.time()) - 40 * 24 * 3600,), + ) + tm._conn.commit() + + assert tm.purge_older_than() == 1 + assert tm.lookup(1) is None + assert tm.lookup(2) == ("222.000", "C0123") + tm.close()