Implement topic support for supergroups

README.md

@@ -67,6 +67,7 @@ All constructor arguments after `token` / `chat_id` are keyword-only.
 |---|---|---|
 | `token` | `None` | Bot token; falls back to `TG_TOKEN`. |
 | `chat_id` | `None` | Target chat; falls back to `TG_CHAT_ID`. |
+| `topic_id` | `None` | Forum topic id; falls back to `TG_TOPIC_ID`. When set on a forum supergroup, messages go to that topic; when unset, to the group's General topic. |
 | `level` | `logging.WARNING` | Minimum level (passed to `setLevel`). |
 | `batch_size` | `1` | Max records per message (`>= 1`). `1` = send each record immediately. |
 | `flush_interval` | `5.0` | Max seconds a partial batch waits (`>= 0`). |
@@ -85,6 +86,7 @@ All constructor arguments after `token` / `chat_id` are keyword-only.
 |---|---|
 | `TG_TOKEN` | `token` argument is omitted. |
 | `TG_CHAT_ID` | `chat_id` argument is omitted. |
+| `TG_TOPIC_ID` | `topic_id` argument is omitted. |
 
 ## Batching
 

tests/test_config_stats.py

@@ -3,11 +3,12 @@
 from __future__ import annotations
 
 import queue
+from typing import cast
 
 import pytest
 
 from tg_logging_handler import queue_policy
-from tg_logging_handler.config import resolve_credentials
+from tg_logging_handler.config import resolve_credentials, resolve_topic_id
 from tg_logging_handler.exceptions import TelegramConfigError
 from tg_logging_handler.stats import HandlerStats, StatsCollector
 
@@ -71,6 +72,48 @@ def test_invalid_long_token_reveals_only_ends(self) -> None:
         assert "mmmm" not in message  # the middle stays hidden
 
 
+class TestResolveTopicId:
+    def test_explicit_int_passes_through(self) -> None:
+        assert resolve_topic_id(42) == 42
+
+    def test_none_returns_none_without_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("TG_TOPIC_ID", raising=False)
+        assert resolve_topic_id(None) is None
+
+    def test_env_string_is_parsed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TG_TOPIC_ID", "42")
+        assert resolve_topic_id(None) == 42
+
+    def test_explicit_arg_wins_over_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TG_TOPIC_ID", "7")
+        assert resolve_topic_id(42) == 42
+
+    def test_empty_env_is_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TG_TOPIC_ID", "")
+        assert resolve_topic_id(None) is None
+
+    def test_invalid_env_string_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TG_TOPIC_ID", "not-a-number")
+        with pytest.raises(ValueError, match="TG_TOPIC_ID must be an integer"):
+            resolve_topic_id(None)
+
+    @pytest.mark.parametrize("bad", [True, False, 3.5, "42"])
+    def test_non_int_explicit_values_are_rejected(self, bad: object) -> None:
+        # bool is an int subclass, floats compare fine, and str would crash on
+        # the range comparison; all must fail with a clean ValueError.
+        with pytest.raises(ValueError, match="topic_id must be an integer"):
+            resolve_topic_id(bad)  # type: ignore[arg-type]  # deliberate misuse
+
+    def test_rejection_message_names_the_type(self) -> None:
+        with pytest.raises(ValueError, match="got bool"):
+            resolve_topic_id(True)
+
+    def test_non_positive_is_rejected(self) -> None:
+        for bad in (0, -1, -100):
+            with pytest.raises(ValueError, match="positive"):
+                resolve_topic_id(bad)
+
+
 class TestStats:
     def test_initial_snapshot_is_zero(self) -> None:
         assert StatsCollector().snapshot() == HandlerStats()
@@ -96,6 +139,31 @@ def test_snapshot_does_not_reflect_later_changes(self) -> None:
         assert stats.snapshot().sent == 1
 
 
+class _ScriptedQueue:
+    """Duck-typed stand-in for ``queue.Queue`` with per-call scripted outcomes.
+
+    Only the two methods ``put_drop_oldest`` touches are provided. ``puts``
+    entries: ``"full"`` makes ``put_nowait`` raise ``queue.Full``, anything
+    else succeeds. ``gets`` entries: ``"empty"`` makes ``get_nowait`` raise
+    ``queue.Empty``, anything else returns an evictable item. This lets tests
+    exercise the concurrency-race branches that a real queue can never reach
+    single-threaded.
+    """
+
+    def __init__(self, puts: list[str], gets: list[str]) -> None:
+        self._puts = iter(puts)
+        self._gets = iter(gets)
+
+    def put_nowait(self, item: object) -> None:
+        if next(self._puts) == "full":
+            raise queue.Full
+
+    def get_nowait(self) -> object:
+        if next(self._gets) == "empty":
+            raise queue.Empty
+        return "evicted"
+
+
 class TestQueuePolicy:
     def test_put_drop_newest_succeeds_when_space(self) -> None:
         q: queue.Queue[object] = queue.Queue(maxsize=1)
@@ -129,6 +197,44 @@ def test_put_drop_oldest_no_eviction_when_space(self) -> None:
         assert result.enqueued is True
         assert result.dropped == 0
 
+    def test_put_drop_oldest_handles_queue_draining_between_evict_and_retry(
+        self,
+    ) -> None:
+        # Race path: between a failed put and the eviction get, another
+        # producer drains the queue, so get_nowait raises Empty and the loop
+        # must retry instead of crashing. Scripted via a fake queue.
+        fake = cast(
+            "queue.Queue[object]",
+            _ScriptedQueue(puts=["full", "full", "ok"], gets=["empty", "item"]),
+        )
+        result = queue_policy.put_drop_oldest(fake, "new")
+        assert result.enqueued is True
+        assert result.dropped == 1  # one real eviction happened, then room appeared
+
+    def test_put_drop_oldest_gives_up_when_slot_keeps_refilling(self) -> None:
+        # Race path: the freed slot is instantly refilled by a competing
+        # producer on every attempt, so all evictions fail and the incoming
+        # item is dropped (and counted) after the bounded attempt budget.
+        fake = cast(
+            "queue.Queue[object]",
+            _ScriptedQueue(puts=["full", "full", "full", "full"], gets=["item"] * 3),
+        )
+        result = queue_policy.put_drop_oldest(fake, "new")
+        assert result.enqueued is False
+        assert result.dropped == 4  # 3 evicted victims + the incoming record
+
+    def test_put_drop_oldest_lands_item_after_attempt_budget(self) -> None:
+        # Race path: the slot keeps refilling for all three attempts, but the
+        # final (post-budget) put succeeds, so the item lands but the victims
+        # are still counted.
+        fake = cast(
+            "queue.Queue[object]",
+            _ScriptedQueue(puts=["full", "full", "full", "ok"], gets=["item"] * 3),
+        )
+        result = queue_policy.put_drop_oldest(fake, "new")
+        assert result.enqueued is True
+        assert result.dropped == 3
+
     def test_put_block_enqueues(self) -> None:
         q: queue.Queue[object] = queue.Queue(maxsize=1)
         result = queue_policy.put_block(q, 1)

tests/test_handler.py

@@ -96,6 +96,58 @@ def test_level_kwarg_applied(fast_handler_factory: HandlerFactory) -> None:
     assert handler.level == logging.ERROR
 
 
+def test_topic_id_validation_raises() -> None:
+    for bad in (0, -1, -100):
+        with pytest.raises(ValueError):
+            TelegramLoggingHandler(
+                token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, topic_id=bad
+            )
+
+
+def test_topic_id_none_is_allowed(fast_handler_factory: HandlerFactory) -> None:
+    handler = fast_handler_factory(topic_id=None)  # None is allowed
+    handler.close()
+
+
+def test_topic_id_from_env_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("TG_TOPIC_ID", "42")
+    handler = TelegramLoggingHandler(
+        token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, api_base_url=API_BASE
+    )
+    assert handler._sender._message_thread_id == 42
+    handler.close()
+
+
+def test_topic_id_arg_wins_over_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("TG_TOPIC_ID", "7")
+    handler = TelegramLoggingHandler(
+        token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, topic_id=42, api_base_url=API_BASE
+    )
+    assert handler._sender._message_thread_id == 42
+    handler.close()
+
+
+def test_topic_id_invalid_env_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("TG_TOPIC_ID", "not-a-number")
+    with pytest.raises(ValueError):
+        TelegramLoggingHandler(
+            token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, api_base_url=API_BASE
+        )
+
+
+def test_topic_id_reaches_the_wire(
+    fast_handler_factory: HandlerFactory, mock_api: respx.MockRouter, wait_for: WaitFor
+) -> None:
+    import json
+
+    handler = fast_handler_factory(topic_id=42)
+    handler.emit(_record("topic log"))
+    assert wait_for(lambda: handler.stats.sent == 1)
+    route = next(r for r in mock_api.routes if "sendMessage" in str(r.pattern))
+    body = json.loads(route.calls[0].request.content.decode())
+    assert body["message_thread_id"] == 42
+
+
 # --- emit contract ----------------------------------------------------------
 
 

tests/test_sender_retry.py

@@ -129,6 +129,64 @@ def test_parse_mode_included_when_set() -> None:
     assert body["parse_mode"] == "HTML"
 
 
+@respx.mock
+def test_message_thread_id_included_when_set() -> None:
+    route = respx.post(SEND_URL).mock(return_value=_ok())
+    sleep = _RecordingSleep()
+    sender = TelegramSender(TOKEN, CHAT, API_BASE, message_thread_id=42, sleep=sleep)
+    sender.send_with_retry("hi")
+    body = json.loads(route.calls[0].request.content.decode())
+    assert body["message_thread_id"] == 42
+
+
+@respx.mock
+def test_message_thread_id_omitted_when_not_set() -> None:
+    route = respx.post(SEND_URL).mock(return_value=_ok())
+    sleep = _RecordingSleep()
+    _sender(sleep).send_with_retry("hi")
+    body = json.loads(route.calls[0].request.content.decode())
+    assert "message_thread_id" not in body  # backward compatible: no topic field
+
+
+@respx.mock
+def test_thread_not_found_400_is_a_permanent_failure() -> None:
+    # Wrong or closed topic id: Telegram answers "message thread not found".
+    # It is a permanent 4xx, so it must not be retried and must count as failed.
+    route = respx.post(SEND_URL).mock(
+        return_value=httpx.Response(400, text="Bad Request: message thread not found")
+    )
+    sleep = _RecordingSleep()
+    outcome = TelegramSender(
+        TOKEN, CHAT, API_BASE, message_thread_id=999, sleep=sleep
+    ).send_with_retry("hi")
+    assert outcome.delivered is False
+    assert outcome.retries == 0
+    assert route.call_count == 1
+    assert sleep.calls == []
+
+
+@respx.mock
+def test_429_heads_up_notice_targets_the_topic() -> None:
+    # The one-time 429 notice must land in the same topic as the logs, not
+    # fall back to the General topic.
+    route = respx.post(SEND_URL).mock(
+        side_effect=[
+            httpx.Response(429, headers={"retry-after": "1"}),
+            _ok(),  # the heads-up notice POST
+            _ok(),  # the retried original message
+        ]
+    )
+    sleep = _RecordingSleep()
+    sender = TelegramSender(TOKEN, CHAT, API_BASE, message_thread_id=42, sleep=sleep)
+    outcome = sender.send_with_retry("real log line")
+    assert outcome.delivered is True
+    notice = [
+        c for c in route.calls if _NOTICE_MARKER in json.loads(c.request.content.decode())["text"]
+    ]
+    assert len(notice) == 1
+    assert json.loads(notice[0].request.content.decode())["message_thread_id"] == 42
+
+
 @respx.mock
 def test_429_waits_are_capped_so_a_stuck_429_cannot_spin_forever() -> None:
     # 11 consecutive 429s (cap is 10) must give up, not loop forever.

tg_logging_handler/config.py

@@ -10,7 +10,7 @@
 from ._constants import DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT
 from .exceptions import TelegramConfigError
 
-__all__ = ["resolve_credentials", "validate_token"]
+__all__ = ["resolve_credentials", "resolve_topic_id", "validate_token"]
 
 
 def resolve_credentials(
@@ -51,6 +51,39 @@ def resolve_credentials(
     return token, str(resolved_chat)
 
 
+def resolve_topic_id(topic_id: int | None) -> int | None:
+    """Resolve the forum topic id from args, falling back to ``TG_TOPIC_ID``.
+
+    Args:
+        topic_id: Explicit topic id, or ``None`` to read ``TG_TOPIC_ID``.
+
+    Returns:
+        A positive ``int`` topic id, or ``None`` when neither is set
+        (messages then go to the group's General topic).
+
+    Raises:
+        ValueError: If the value is not an integer or not positive.
+    """
+    if topic_id is not None:
+        if type(topic_id) is not int:
+            # bool is an int subclass and floats compare fine, but both are
+            # always mistakes here; anything else (e.g. str) would crash on
+            # the range comparison below. Reject non-ints up front.
+            raise ValueError(f"topic_id must be an integer, got {type(topic_id).__name__}")
+        value = topic_id
+    else:
+        raw = os.environ.get("TG_TOPIC_ID")
+        if raw is None or raw == "":
+            return None
+        try:
+            value = int(raw)
+        except ValueError as exc:
+            raise ValueError(f"TG_TOPIC_ID must be an integer, got {raw!r}") from exc
+    if value < 1:
+        raise ValueError("topic_id must be a positive integer")
+    return value
+
+
 def _looks_like_bot_token(token: str) -> bool:
     """Heuristic: Telegram bot tokens are ``'<numeric bot_id>:<auth>'``."""
     return re.fullmatch(r"\d+:.+", token) is not None

tg_logging_handler/handler.py

@@ -14,7 +14,7 @@
 from typing import Literal
 
 from . import queue_policy
-from .config import resolve_credentials, validate_token
+from .config import resolve_credentials, resolve_topic_id, validate_token
 from .sender import TelegramSender
 from .stats import HandlerStats, StatsCollector
 from .worker import SHUTDOWN, WorkerThread
@@ -31,6 +31,9 @@ class TelegramLoggingHandler(logging.Handler):
     Args:
         token: Bot token. Falls back to ``TG_TOKEN``. Raises if neither is set.
         chat_id: Target chat id. Falls back to ``TG_CHAT_ID``. Raises if neither is set.
+        topic_id: Target forum topic id. Falls back to ``TG_TOPIC_ID``. When set
+            and the chat is a forum supergroup, messages go to that topic; when
+            unset they go to the group's General topic.
         level: Minimum level, passed to ``setLevel``. Defaults to ``WARNING``.
         batch_size: Max records per outgoing message (``>= 1``).
         flush_interval: Max seconds a partial batch waits (``>= 0``).
@@ -52,6 +55,7 @@ def __init__(
         token: str | None = None,
         chat_id: str | int | None = None,
         *,
+        topic_id: int | None = None,
         level: int | str = logging.WARNING,
         batch_size: int = 1,
         flush_interval: float = 5.0,
@@ -76,6 +80,7 @@ def __init__(
             raise ValueError("queue_maxsize must be >= 0")
 
         resolved_token, resolved_chat = resolve_credentials(token, chat_id)
+        resolved_topic = resolve_topic_id(topic_id)
         if validate:
             validate_token(resolved_token, api_base_url)
 
@@ -96,6 +101,7 @@ def __init__(
             api_base_url,
             parse_mode=parse_mode,
             max_retries=max_retries,
+            message_thread_id=resolved_topic,
         )
         self._worker = WorkerThread(
             record_queue=self._queue,