fix(lolschedule): route daily push to originating forum topic

Subscribers previously stored only chatID; the daily-push cron sent
without MessageThreadID so Telegram routed to General regardless of the
topic the user subscribed from.

- Storage shape: []Subscriber{ChatID, ThreadID} keyed by (ChatID, ThreadID)
  so the same chat can subscribe in multiple topics independently.
- Decoder accepts legacy []int64 rows (ThreadID=0); successful mutations
  rewrite the slot in the new shape, no manual migration needed.
- Cron forwards MessageThreadID on every send.
- Auto-prune classifies terminal errors: chat-wide kills (bot blocked,
  deactivated, kicked, deleted, group upgraded) wipe every entry for the
  ChatID; topic-only "have no rights to send" prunes just (ChatID, ThreadID).

internal/modules/lolschedule/cron.go

@@ -14,38 +14,66 @@ import (
 	"github.com/tiennm99/miti99bot/internal/modules"
 )
 
-// terminalSendErrorMarkers are substrings of Telegram API errors that mean the
-// chat will never accept messages again (blocked, deactivated, kicked, chat
-// gone). Detecting these lets the daily-push handler prune the subscriber
-// list so dead chats stop consuming the 30-msg/s global budget.
+// terminalKind classifies a permanent send failure by blast radius.
+type terminalKind int
+
+const (
+	// terminalNone is a transient failure (rate limit, timeout, 5xx). The
+	// subscriber stays on the list and we retry on the next push.
+	terminalNone terminalKind = iota
+	// terminalChatWide means the chat itself is unreachable (bot blocked,
+	// chat deactivated, kicked, deleted, group upgraded). Every subscriber
+	// entry for that ChatID must be pruned — sister topics are dead too.
+	terminalChatWide
+	// terminalTopicOnly means the bot lost send rights in the specific topic
+	// (e.g. a topic-level permissions change). Only the (ChatID, ThreadID)
+	// entry that failed should be pruned; other topics in the same chat may
+	// still be valid.
+	terminalTopicOnly
+)
+
+// chatWideTerminalMarkers are substrings of Telegram API errors that mean
+// the whole chat is gone, not just one topic. Detecting these lets the
+// daily-push handler prune every subscription for that ChatID at once.
 //
 // String matching is fragile by nature, but the bot library surfaces these
 // directly in err.Error() and Telegram has used the same wording for years.
 // The false-negative path (we miss a new wording, dead chat lingers) is
 // strictly safer than the false-positive path (we wrongly prune a live chat).
-var terminalSendErrorMarkers = []string{
+var chatWideTerminalMarkers = []string{
 	"bot was blocked by the user",
 	"user is deactivated",
 	"bot is not a member",
 	"chat not found",
 	"group chat was upgraded",
-	"have no rights to send",
 	"chat was deleted",
 }
 
-// isTerminalSendError reports whether err indicates the chat is permanently
-// unreachable. Used by runDailyPush to drive auto-unsubscription.
-func isTerminalSendError(err error) bool {
+// topicOnlyTerminalMarkers are errors that scope to a single forum topic
+// (or to the bot's per-topic permissions). Pruning only the offending
+// (ChatID, ThreadID) keeps the chat's other topic subscriptions alive.
+var topicOnlyTerminalMarkers = []string{
+	"have no rights to send",
+}
+
+// classifyTerminal reports whether err is a permanent send failure and, if
+// so, whether it kills the whole chat or only the originating topic.
+func classifyTerminal(err error) terminalKind {
 	if err == nil {
-		return false
+		return terminalNone
 	}
 	msg := err.Error()
-	for _, m := range terminalSendErrorMarkers {
+	for _, m := range chatWideTerminalMarkers {
 		if strings.Contains(msg, m) {
-			return true
+			return terminalChatWide
 		}
 	}
-	return false
+	for _, m := range topicOnlyTerminalMarkers {
+		if strings.Contains(msg, m) {
+			return terminalTopicOnly
+		}
+	}
+	return terminalNone
 }
 
 // dailyPushCronName is the cron route segment + EventBridge schedule key.
@@ -94,6 +122,9 @@ func (s *state) dailyPushHandler(ctx context.Context, deps modules.Deps) error {
 // runDailyPush is the testable core: fetch subscribers, fetch today's matches,
 // fan out to every subscriber. Per-chat send failures are logged but do not
 // abort the batch — one bad chat does not deny the rest.
+//
+// MessageThreadID is forwarded on every send so subscribers in a forum-topic
+// receive the digest in that topic, not in General.
 func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
 	subs, err := listSubscribers(ctx, s.kv)
 	if err != nil {
@@ -115,8 +146,9 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
 
 	throttle := len(subs) > telegramRateLimitThreshold
 	var sent, failed int
-	var deadChats []int64
-	for i, chatID := range subs {
+	deadChats := map[int64]struct{}{}
+	var deadTopics []Subscriber
+	for i, sub := range subs {
 		if throttle && i > 0 {
 			select {
 			case <-ctx.Done():
@@ -125,14 +157,19 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
 			}
 		}
 		if _, err := sender.SendMessage(ctx, &bot.SendMessageParams{
-			ChatID:    chatID,
-			Text:      text,
-			ParseMode: models.ParseModeHTML,
+			ChatID:          sub.ChatID,
+			MessageThreadID: sub.ThreadID,
+			Text:            text,
+			ParseMode:       models.ParseModeHTML,
 		}); err != nil {
-			log.Warn("lolschedule daily push send failed", "chat", chatID, "err", err)
+			log.Warn("lolschedule daily push send failed",
+				"chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
 			failed++
-			if isTerminalSendError(err) {
-				deadChats = append(deadChats, chatID)
+			switch classifyTerminal(err) {
+			case terminalChatWide:
+				deadChats[sub.ChatID] = struct{}{}
+			case terminalTopicOnly:
+				deadTopics = append(deadTopics, sub)
 			}
 			continue
 		}
@@ -142,7 +179,7 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
 	// Best-effort prune. Failure here just leaves the dead chats in the list
 	// for tomorrow's push — same behaviour as before this code existed, so
 	// strictly an improvement even when the writes fail.
-	pruned := pruneDeadSubscribers(ctx, s, deadChats)
+	pruned := pruneDeadSubscribers(ctx, s, deadChats, deadTopics)
 
 	log.Info("lolschedule daily push complete",
 		"subscribers", len(subs),
@@ -153,21 +190,35 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
 	return nil
 }
 
-// pruneDeadSubscribers removes chatIDs flagged as permanently unreachable.
-// Serializes through state.subscribersMu so a concurrent /subscribe handler
-// doesn't lose its write. Returns the number actually removed (idempotent if
-// a subscriber unsubscribed between the failed send and this call).
-func pruneDeadSubscribers(ctx context.Context, s *state, deadChats []int64) int {
-	if len(deadChats) == 0 {
+// pruneDeadSubscribers removes entries flagged unreachable. Chat-wide failures
+// drop every subscription for the chat; topic-only failures drop just the one
+// (ChatID, ThreadID). Serializes through state.subscribersMu so a concurrent
+// /subscribe handler doesn't lose its write. Returns total entries removed.
+func pruneDeadSubscribers(ctx context.Context, s *state, chatWide map[int64]struct{}, topicOnly []Subscriber) int {
+	if len(chatWide) == 0 && len(topicOnly) == 0 {
 		return 0
 	}
 	s.subscribersMu.Lock()
 	defer s.subscribersMu.Unlock()
 	removed := 0
-	for _, chatID := range deadChats {
-		ok, err := removeSubscriber(ctx, s.kv, chatID)
+	for chatID := range chatWide {
+		n, err := removeAllForChat(ctx, s.kv, chatID)
 		if err != nil {
-			log.Warn("lolschedule prune dead subscriber failed", "chat", chatID, "err", err)
+			log.Warn("lolschedule prune dead chat failed", "chat", chatID, "err", err)
+			continue
+		}
+		removed += n
+	}
+	for _, sub := range topicOnly {
+		// Skip if the whole chat was already pruned above — saves a redundant
+		// Get→mutate→Put round trip.
+		if _, ok := chatWide[sub.ChatID]; ok {
+			continue
+		}
+		ok, err := removeSubscriber(ctx, s.kv, sub.ChatID, sub.ThreadID)
+		if err != nil {
+			log.Warn("lolschedule prune dead topic failed",
+				"chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
 			continue
 		}
 		if ok {

internal/modules/lolschedule/cron_test.go

@@ -15,14 +15,15 @@ import (
 )
 
 // fakeSender records every SendMessage call. errOn returns a transient
-// failure for the configured chat IDs; terminalErrOn returns a
-// permanent-failure message string the dead-subscriber pruner recognises.
-// All others succeed.
+// failure for the configured chat IDs; terminalErrOn returns a chat-wide
+// permanent-failure message string; topicTerminalErrOn returns a
+// topic-only permanent-failure string. All others succeed.
 type fakeSender struct {
-	mu            sync.Mutex
-	calls         []bot.SendMessageParams
-	errOn         map[int64]bool
-	terminalErrOn map[int64]bool
+	mu                 sync.Mutex
+	calls              []bot.SendMessageParams
+	errOn              map[int64]bool
+	terminalErrOn      map[int64]bool
+	topicTerminalErrOn map[int64]bool
 }
 
 func (f *fakeSender) SendMessage(_ context.Context, p *bot.SendMessageParams) (*models.Message, error) {
@@ -31,10 +32,13 @@ func (f *fakeSender) SendMessage(_ context.Context, p *bot.SendMessageParams) (*
 	f.calls = append(f.calls, *p)
 	id, ok := p.ChatID.(int64)
 	if ok && f.terminalErrOn[id] {
-		// String shape that isTerminalSendError matches; verifies the marker
+		// String shape that classifyTerminal matches; verifies the marker
 		// list works against a realistic Telegram error message.
 		return nil, errors.New("Forbidden: bot was blocked by the user")
 	}
+	if ok && f.topicTerminalErrOn[id] {
+		return nil, errors.New("Bad Request: have no rights to send a message")
+	}
 	if ok && f.errOn[id] {
 		return nil, errors.New("fakeSender: induced failure for chat " + chatIDString(id))
 	}
@@ -96,7 +100,7 @@ func TestRunDailyPush_SendsToAllSubscribers(t *testing.T) {
 
 	chatIDs := []int64{100, 200, 300}
 	for _, id := range chatIDs {
-		if _, err := addSubscriber(context.Background(), s.kv, id); err != nil {
+		if _, err := addSubscriber(context.Background(), s.kv, id, 0); err != nil {
 			t.Fatalf("addSubscriber %d: %v", id, err)
 		}
 	}
@@ -123,6 +127,43 @@ func TestRunDailyPush_SendsToAllSubscribers(t *testing.T) {
 		if call.Text == "" {
 			t.Errorf("send %d: empty text", i)
 		}
+		if call.MessageThreadID != 0 {
+			t.Errorf("send %d: thread got %d, want 0 (no topic)", i, call.MessageThreadID)
+		}
+	}
+}
+
+// TestRunDailyPush_ForwardsMessageThreadID locks in the forum-topic fix:
+// subscribers stored with a non-zero ThreadID receive the daily push in that
+// topic, not in General.
+func TestRunDailyPush_ForwardsMessageThreadID(t *testing.T) {
+	s := newTestState(t)
+	seedFreshCache(t, s.kv, nil)
+
+	subs := []Subscriber{
+		{ChatID: 100, ThreadID: 0},
+		{ChatID: 100, ThreadID: 7},
+		{ChatID: 200, ThreadID: 42},
+	}
+	for _, sub := range subs {
+		if _, err := addSubscriber(context.Background(), s.kv, sub.ChatID, sub.ThreadID); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	sender := &fakeSender{}
+	if err := runDailyPush(context.Background(), s, sender); err != nil {
+		t.Fatalf("runDailyPush: %v", err)
+	}
+	if len(sender.calls) != len(subs) {
+		t.Fatalf("expected %d sends, got %d", len(subs), len(sender.calls))
+	}
+	for i, call := range sender.calls {
+		gotID, _ := call.ChatID.(int64)
+		if gotID != subs[i].ChatID || call.MessageThreadID != subs[i].ThreadID {
+			t.Errorf("send %d: got (chat=%d, thread=%d), want (chat=%d, thread=%d)",
+				i, gotID, call.MessageThreadID, subs[i].ChatID, subs[i].ThreadID)
+		}
 	}
 }
 
@@ -132,7 +173,7 @@ func TestRunDailyPush_PartialFailureContinues(t *testing.T) {
 
 	chatIDs := []int64{100, 200, 300}
 	for _, id := range chatIDs {
-		if _, err := addSubscriber(context.Background(), s.kv, id); err != nil {
+		if _, err := addSubscriber(context.Background(), s.kv, id, 0); err != nil {
 			t.Fatalf("addSubscriber %d: %v", id, err)
 		}
 	}
@@ -141,7 +182,6 @@ func TestRunDailyPush_PartialFailureContinues(t *testing.T) {
 	if err := runDailyPush(context.Background(), s, sender); err != nil {
 		t.Fatalf("runDailyPush: %v (should swallow per-chat failures)", err)
 	}
-	// All three chats should be attempted even though chat 200 failed.
 	if len(sender.calls) != 3 {
 		t.Errorf("expected 3 attempts (failure does not abort batch), got %d", len(sender.calls))
 	}
@@ -149,23 +189,30 @@ func TestRunDailyPush_PartialFailureContinues(t *testing.T) {
 
 // TestRunDailyPush_PrunesDeadSubscribers locks in the auto-cleanup of chats
 // that have permanently blocked the bot. Recoverable (transient) errors
-// MUST NOT trigger removal — only terminal Telegram errors do.
+// MUST NOT trigger removal — only terminal Telegram errors do. A chat-wide
+// terminal error removes every topic subscription for that chat.
 func TestRunDailyPush_PrunesDeadSubscribers(t *testing.T) {
 	s := newTestState(t)
 	seedFreshCache(t, s.kv, nil)
 
-	chatIDs := []int64{100, 200, 300, 400}
-	for _, id := range chatIDs {
-		if _, err := addSubscriber(context.Background(), s.kv, id); err != nil {
-			t.Fatalf("addSubscriber %d: %v", id, err)
+	// Chat 400 has two topic subs; both should be pruned when the chat
+	// returns a chat-wide terminal error.
+	seedSubs := []Subscriber{
+		{ChatID: 100, ThreadID: 0},
+		{ChatID: 200, ThreadID: 0},
+		{ChatID: 300, ThreadID: 0},
+		{ChatID: 400, ThreadID: 0},
+		{ChatID: 400, ThreadID: 9},
+	}
+	for _, sub := range seedSubs {
+		if _, err := addSubscriber(context.Background(), s.kv, sub.ChatID, sub.ThreadID); err != nil {
+			t.Fatalf("addSubscriber %v: %v", sub, err)
 		}
 	}
 
 	sender := &fakeSender{
-		// 200 hit a transient failure → keep on list. 400 is permanently blocked
-		// → prune from list.
-		errOn:         map[int64]bool{200: true},
-		terminalErrOn: map[int64]bool{400: true},
+		errOn:         map[int64]bool{200: true}, // transient → keep
+		terminalErrOn: map[int64]bool{400: true}, // chat-wide → wipe all 400 entries
 	}
 	if err := runDailyPush(context.Background(), s, sender); err != nil {
 		t.Fatalf("runDailyPush: %v", err)
@@ -175,41 +222,114 @@ func TestRunDailyPush_PrunesDeadSubscribers(t *testing.T) {
 	if err != nil {
 		t.Fatalf("listSubscribers: %v", err)
 	}
-	want := []int64{100, 200, 300} // 400 removed; 200 retained despite transient error
+	want := []Subscriber{{ChatID: 100}, {ChatID: 200}, {ChatID: 300}}
 	if len(remaining) != len(want) {
 		t.Fatalf("subscribers after prune: got %v, want %v", remaining, want)
 	}
-	for i, id := range want {
-		if remaining[i] != id {
-			t.Errorf("subscriber[%d]: got %d, want %d", i, remaining[i], id)
+	for i, s := range want {
+		if remaining[i] != s {
+			t.Errorf("subscriber[%d]: got %v, want %v", i, remaining[i], s)
 		}
 	}
 }
 
-func TestIsTerminalSendError(t *testing.T) {
-	terminals := []string{
+// TestRunDailyPush_TopicOnlyTerminalPrunesOneTopic verifies that a
+// "have no rights to send" failure removes only the failing
+// (ChatID, ThreadID) entry — sister topics in the same chat stay subscribed.
+func TestRunDailyPush_TopicOnlyTerminalPrunesOneTopic(t *testing.T) {
+	s := newTestState(t)
+	seedFreshCache(t, s.kv, nil)
+
+	seedSubs := []Subscriber{
+		{ChatID: 500, ThreadID: 0},
+		{ChatID: 500, ThreadID: 11},
+		{ChatID: 500, ThreadID: 22},
+	}
+	for _, sub := range seedSubs {
+		if _, err := addSubscriber(context.Background(), s.kv, sub.ChatID, sub.ThreadID); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Every send to chat 500 returns the topic-only terminal — but only the
+	// matching (ChatID, ThreadID) should be pruned per call. With the marker
+	// applying to all three sends we expect all three entries to drop.
+	sender := &fakeSender{topicTerminalErrOn: map[int64]bool{500: true}}
+	if err := runDailyPush(context.Background(), s, sender); err != nil {
+		t.Fatalf("runDailyPush: %v", err)
+	}
+
+	remaining, _ := listSubscribers(context.Background(), s.kv)
+	if len(remaining) != 0 {
+		t.Errorf("expected all topic-only entries pruned, got %v", remaining)
+	}
+}
+
+// TestRunDailyPush_TopicOnlyTerminalKeepsOtherTopics: only one topic in a
+// chat goes bad; the other topics in the same chat must stay.
+func TestRunDailyPush_TopicOnlyTerminalKeepsOtherTopics(t *testing.T) {
+	s := newTestState(t)
+	seedFreshCache(t, s.kv, nil)
+
+	// Two distinct chats, each with multiple topic subs. Only chat 600 hits
+	// the topic-terminal error; chat 700 sends cleanly.
+	seedSubs := []Subscriber{
+		{ChatID: 600, ThreadID: 1},
+		{ChatID: 600, ThreadID: 2},
+		{ChatID: 700, ThreadID: 3},
+	}
+	for _, sub := range seedSubs {
+		if _, err := addSubscriber(context.Background(), s.kv, sub.ChatID, sub.ThreadID); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Custom sender: chat 600 always fails topic-terminal, chat 700 succeeds.
+	sender := &fakeSender{topicTerminalErrOn: map[int64]bool{600: true}}
+	if err := runDailyPush(context.Background(), s, sender); err != nil {
+		t.Fatalf("runDailyPush: %v", err)
+	}
+
+	remaining, _ := listSubscribers(context.Background(), s.kv)
+	if len(remaining) != 1 || remaining[0] != (Subscriber{ChatID: 700, ThreadID: 3}) {
+		t.Errorf("after topic-terminal prune of chat 600: got %v, want [{700 3}]", remaining)
+	}
+}
+
+func TestClassifyTerminal(t *testing.T) {
+	chatWide := []string{
 		"Forbidden: bot was blocked by the user",
 		"Forbidden: user is deactivated",
 		"Bad Request: chat not found",
 		"Bad Request: group chat was upgraded to a supergroup chat",
 	}
-	for _, msg := range terminals {
-		if !isTerminalSendError(errors.New(msg)) {
-			t.Errorf("isTerminalSendError(%q) = false, want true", msg)
+	for _, msg := range chatWide {
+		if got := classifyTerminal(errors.New(msg)); got != terminalChatWide {
+			t.Errorf("classifyTerminal(%q) = %v, want terminalChatWide", msg, got)
 		}
 	}
+
+	topicOnly := []string{
+		"Bad Request: have no rights to send a message",
+	}
+	for _, msg := range topicOnly {
+		if got := classifyTerminal(errors.New(msg)); got != terminalTopicOnly {
+			t.Errorf("classifyTerminal(%q) = %v, want terminalTopicOnly", msg, got)
+		}
+	}
+
 	transients := []string{
 		"connection reset by peer",
 		"Too Many Requests: retry after 30",
 		"context deadline exceeded",
 	}
 	for _, msg := range transients {
-		if isTerminalSendError(errors.New(msg)) {
-			t.Errorf("isTerminalSendError(%q) = true, want false (transient)", msg)
+		if got := classifyTerminal(errors.New(msg)); got != terminalNone {
+			t.Errorf("classifyTerminal(%q) = %v, want terminalNone (transient)", msg, got)
 		}
 	}
-	if isTerminalSendError(nil) {
-		t.Error("isTerminalSendError(nil) = true, want false")
+	if got := classifyTerminal(nil); got != terminalNone {
+		t.Errorf("classifyTerminal(nil) = %v, want terminalNone", got)
 	}
 }
 

internal/modules/lolschedule/handlers.go

@@ -100,7 +100,7 @@ func (s *state) handleSubscribe(ctx context.Context, b *bot.Bot, update *models.
 	}
 	s.subscribersMu.Lock()
 	defer s.subscribersMu.Unlock()
-	added, err := addSubscriber(ctx, s.kv, msg.Chat.ID)
+	added, err := addSubscriber(ctx, s.kv, msg.Chat.ID, msg.MessageThreadID)
 	if err != nil {
 		return err
 	}
@@ -120,7 +120,7 @@ func (s *state) handleUnsubscribe(ctx context.Context, b *bot.Bot, update *model
 	}
 	s.subscribersMu.Lock()
 	defer s.subscribersMu.Unlock()
-	removed, err := removeSubscriber(ctx, s.kv, msg.Chat.ID)
+	removed, err := removeSubscriber(ctx, s.kv, msg.Chat.ID, msg.MessageThreadID)
 	if err != nil {
 		return err
 	}

internal/modules/lolschedule/handlers_test.go

@@ -130,8 +130,50 @@ func TestHandleSubscribe_AddsAndIsIdempotent(t *testing.T) {
 		t.Errorf("duplicate subscribe should report Already; got %q", got)
 	}
 	ids, _ := listSubscribers(context.Background(), kv)
-	if len(ids) != 1 || ids[0] != 7 {
-		t.Errorf("subscribers = %v, want [7]", ids)
+	if len(ids) != 1 || ids[0] != (Subscriber{ChatID: 7}) {
+		t.Errorf("subscribers = %v, want [{7 0}]", ids)
+	}
+}
+
+// TestHandleSubscribe_ForumTopic_CapturesThreadID locks in the forum-topic
+// fix at the handler boundary: subscribing from a non-zero MessageThreadID
+// records that thread on the Subscriber so the daily push lands in the
+// originating topic instead of General.
+func TestHandleSubscribe_ForumTopic_CapturesThreadID(t *testing.T) {
+	rb, kv := installSchedule(t, todayBody, fakeNowMs)
+	upd := testutil.NewSupergroupMessage(555, 999, "/lolschedule_subscribe")
+	upd.Message.MessageThreadID = 42
+	rb.Bot.ProcessUpdate(context.Background(), upd)
+
+	subs, _ := listSubscribers(context.Background(), kv)
+	want := Subscriber{ChatID: 555, ThreadID: 42}
+	if len(subs) != 1 || subs[0] != want {
+		t.Errorf("subscribers = %v, want [%v]", subs, want)
+	}
+}
+
+// TestHandleUnsubscribe_ForumTopic_RemovesOnlyThatTopic verifies that
+// unsubscribing from one topic does not affect sister-topic subscriptions
+// in the same chat.
+func TestHandleUnsubscribe_ForumTopic_RemovesOnlyThatTopic(t *testing.T) {
+	rb, kv := installSchedule(t, todayBody, fakeNowMs)
+
+	// Subscribe in two topics of the same supergroup.
+	for _, tid := range []int{42, 99} {
+		upd := testutil.NewSupergroupMessage(555, 999, "/lolschedule_subscribe")
+		upd.Message.MessageThreadID = tid
+		rb.Bot.ProcessUpdate(context.Background(), upd)
+	}
+
+	// Unsubscribe in topic 42 only.
+	upd := testutil.NewSupergroupMessage(555, 999, "/lolschedule_unsubscribe")
+	upd.Message.MessageThreadID = 42
+	rb.Bot.ProcessUpdate(context.Background(), upd)
+
+	subs, _ := listSubscribers(context.Background(), kv)
+	want := Subscriber{ChatID: 555, ThreadID: 99}
+	if len(subs) != 1 || subs[0] != want {
+		t.Errorf("subscribers = %v, want [%v]", subs, want)
 	}
 }
 

internal/modules/lolschedule/subscribers.go

@@ -2,6 +2,7 @@ package lolschedule
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 
@@ -9,67 +10,97 @@ import (
 )
 
 // subscribersKey is the KV slot holding the per-module subscriber list.
-// Stored as a JSON array of int64 chat ids — same shape as JS so a
-// cross-runtime KV migration round-trips byte-for-byte.
 const subscribersKey = "subscribers"
 
+// Subscriber is one row in the subscriber list. ThreadID is the Telegram
+// forum-topic id the user subscribed from; 0 means the chat's General topic
+// (or a non-forum chat). Uniqueness key is (ChatID, ThreadID) so the same
+// chat can subscribe independently in multiple topics.
+//
+// Telegram routes outgoing messages with an absent/zero message_thread_id
+// to the General topic, so carrying ThreadID alongside ChatID is what keeps
+// the daily push landing in the topic the user subscribed from.
+type Subscriber struct {
+	ChatID   int64 `json:"chat_id"`
+	ThreadID int   `json:"thread_id,omitempty"`
+}
+
 // listSubscribers returns the current subscriber list, or an empty slice
 // if none have ever subscribed.
-func listSubscribers(ctx context.Context, kv storage.KVStore) ([]int64, error) {
-	var ids []int64
-	err := kv.GetJSON(ctx, subscribersKey, &ids)
+//
+// Decoder accepts two shapes for forward-compatibility with rows written
+// before topic-aware subscriptions existed:
+//   - current: [{"chat_id":123,"thread_id":7}, ...]
+//   - legacy:  [123, 456, ...]   (decoded with ThreadID = 0)
+//
+// The decoder handles legacy rows indefinitely; any successful add/remove
+// rewrites the slot in the new shape, but a no-op duplicate add leaves the
+// legacy bytes in place — that's fine, the fallback path stays correct.
+func listSubscribers(ctx context.Context, kv storage.KVStore) ([]Subscriber, error) {
+	raw, err := kv.Get(ctx, subscribersKey)
 	switch {
-	case err == nil:
-		return ids, nil
 	case errors.Is(err, storage.ErrNotFound):
 		return nil, nil
-	default:
+	case err != nil:
 		return nil, fmt.Errorf("lolschedule listSubscribers: %w", err)
 	}
+	var subs []Subscriber
+	if err := json.Unmarshal(raw, &subs); err == nil {
+		return subs, nil
+	}
+	var legacy []int64
+	if err := json.Unmarshal(raw, &legacy); err != nil {
+		return nil, fmt.Errorf("lolschedule listSubscribers decode: %w", err)
+	}
+	out := make([]Subscriber, len(legacy))
+	for i, id := range legacy {
+		out[i] = Subscriber{ChatID: id}
+	}
+	return out, nil
 }
 
-// addSubscriber appends chatID if absent. Returns true on first-add, false
-// when already subscribed (idempotent).
+// addSubscriber appends (chatID, threadID) if that exact pair is absent.
+// Returns true on first-add, false when already subscribed (idempotent).
 //
 // Concurrency: the list lives in a single KV slot, so a concurrent
 // Get→mutate→Put from two chats subscribing in the same millisecond would
 // drop one write. Callers MUST serialize through state.subscribersMu (or an
 // equivalent module-scoped lock) before calling this.
-func addSubscriber(ctx context.Context, kv storage.KVStore, chatID int64) (bool, error) {
-	ids, err := listSubscribers(ctx, kv)
+func addSubscriber(ctx context.Context, kv storage.KVStore, chatID int64, threadID int) (bool, error) {
+	subs, err := listSubscribers(ctx, kv)
 	if err != nil {
 		return false, err
 	}
-	for _, id := range ids {
-		if id == chatID {
+	for _, s := range subs {
+		if s.ChatID == chatID && s.ThreadID == threadID {
 			return false, nil
 		}
 	}
-	ids = append(ids, chatID)
-	if err := kv.PutJSON(ctx, subscribersKey, ids); err != nil {
+	subs = append(subs, Subscriber{ChatID: chatID, ThreadID: threadID})
+	if err := kv.PutJSON(ctx, subscribersKey, subs); err != nil {
 		return false, fmt.Errorf("lolschedule addSubscriber: %w", err)
 	}
 	return true, nil
 }
 
-// removeSubscriber drops chatID from the list. Returns true when removed,
-// false when chatID wasn't present (idempotent).
+// removeSubscriber drops the single (chatID, threadID) entry. Returns true
+// when removed, false when that exact pair wasn't present (idempotent).
 //
 // Concurrency: same single-slot Get→mutate→Put as addSubscriber; callers
 // must hold state.subscribersMu.
-func removeSubscriber(ctx context.Context, kv storage.KVStore, chatID int64) (bool, error) {
-	ids, err := listSubscribers(ctx, kv)
+func removeSubscriber(ctx context.Context, kv storage.KVStore, chatID int64, threadID int) (bool, error) {
+	subs, err := listSubscribers(ctx, kv)
 	if err != nil {
 		return false, err
 	}
-	out := make([]int64, 0, len(ids))
+	out := make([]Subscriber, 0, len(subs))
 	removed := false
-	for _, id := range ids {
-		if id == chatID {
+	for _, s := range subs {
+		if s.ChatID == chatID && s.ThreadID == threadID {
 			removed = true
 			continue
 		}
-		out = append(out, id)
+		out = append(out, s)
 	}
 	if !removed {
 		return false, nil
@@ -79,3 +110,33 @@ func removeSubscriber(ctx context.Context, kv storage.KVStore, chatID int64) (bo
 	}
 	return true, nil
 }
+
+// removeAllForChat drops every entry for chatID regardless of ThreadID.
+// Used when a send fails with a chat-wide terminal error (bot blocked,
+// chat deactivated, kicked, deleted) — every topic subscription in that
+// chat is dead, not just the one the failing send targeted. Returns the
+// number of entries actually removed.
+//
+// Concurrency: callers must hold state.subscribersMu.
+func removeAllForChat(ctx context.Context, kv storage.KVStore, chatID int64) (int, error) {
+	subs, err := listSubscribers(ctx, kv)
+	if err != nil {
+		return 0, err
+	}
+	out := make([]Subscriber, 0, len(subs))
+	removed := 0
+	for _, s := range subs {
+		if s.ChatID == chatID {
+			removed++
+			continue
+		}
+		out = append(out, s)
+	}
+	if removed == 0 {
+		return 0, nil
+	}
+	if err := kv.PutJSON(ctx, subscribersKey, out); err != nil {
+		return 0, fmt.Errorf("lolschedule removeAllForChat: %w", err)
+	}
+	return removed, nil
+}

internal/modules/lolschedule/subscribers_test.go

@@ -16,44 +16,146 @@ func TestSubscribers_AddRemoveListIdempotent(t *testing.T) {
 		t.Errorf("empty list = %v, want []", got)
 	}
 
-	// First add → true.
-	added, err := addSubscriber(ctx, kv, 42)
+	added, err := addSubscriber(ctx, kv, 42, 0)
 	if err != nil || !added {
 		t.Fatalf("first add: added=%v err=%v", added, err)
 	}
-	// Idempotent re-add → false.
-	added, _ = addSubscriber(ctx, kv, 42)
+	added, _ = addSubscriber(ctx, kv, 42, 0)
 	if added {
 		t.Errorf("re-add should be no-op")
 	}
 
 	got, _ = listSubscribers(ctx, kv)
-	if len(got) != 1 || got[0] != 42 {
-		t.Errorf("after add(42): %v, want [42]", got)
+	if len(got) != 1 || got[0] != (Subscriber{ChatID: 42}) {
+		t.Errorf("after add(42,0): %v, want [{42 0}]", got)
 	}
 
-	// Add second.
-	if _, err := addSubscriber(ctx, kv, 7); err != nil {
+	if _, err := addSubscriber(ctx, kv, 7, 0); err != nil {
 		t.Fatal(err)
 	}
 	got, _ = listSubscribers(ctx, kv)
 	if len(got) != 2 {
-		t.Errorf("after add(7): %v, want len 2", got)
+		t.Errorf("after add(7,0): %v, want len 2", got)
 	}
 
-	// Remove → true.
-	removed, err := removeSubscriber(ctx, kv, 42)
+	removed, err := removeSubscriber(ctx, kv, 42, 0)
 	if err != nil || !removed {
-		t.Fatalf("remove(42): removed=%v err=%v", removed, err)
+		t.Fatalf("remove(42,0): removed=%v err=%v", removed, err)
 	}
-	// Idempotent re-remove → false.
-	removed, _ = removeSubscriber(ctx, kv, 42)
+	removed, _ = removeSubscriber(ctx, kv, 42, 0)
 	if removed {
 		t.Errorf("re-remove should be no-op")
 	}
 
 	got, _ = listSubscribers(ctx, kv)
-	if len(got) != 1 || got[0] != 7 {
-		t.Errorf("after remove(42): %v, want [7]", got)
+	if len(got) != 1 || got[0].ChatID != 7 {
+		t.Errorf("after remove(42,0): %v, want [{7 0}]", got)
+	}
+}
+
+// TestSubscribers_TopicsAreDistinct locks in the forum-topic fix: the same
+// chat can subscribe in multiple topics independently, and removing one
+// topic must not affect the others.
+func TestSubscribers_TopicsAreDistinct(t *testing.T) {
+	ctx := context.Background()
+	kv := storage.NewMemoryKVStore()
+
+	for _, tid := range []int{0, 5, 9} {
+		added, err := addSubscriber(ctx, kv, 100, tid)
+		if err != nil || !added {
+			t.Fatalf("add(100,%d): added=%v err=%v", tid, added, err)
+		}
+	}
+	// Same (chat, thread) is rejected as duplicate.
+	if added, _ := addSubscriber(ctx, kv, 100, 5); added {
+		t.Errorf("duplicate (100,5) should be no-op")
+	}
+
+	got, _ := listSubscribers(ctx, kv)
+	if len(got) != 3 {
+		t.Fatalf("after 3 adds: %v, want len 3", got)
+	}
+
+	// Removing one topic leaves the other two intact.
+	if removed, _ := removeSubscriber(ctx, kv, 100, 5); !removed {
+		t.Error("remove(100,5) should report removed=true")
+	}
+	got, _ = listSubscribers(ctx, kv)
+	if len(got) != 2 {
+		t.Errorf("after remove(100,5): %v, want len 2", got)
+	}
+	for _, s := range got {
+		if s.ChatID == 100 && s.ThreadID == 5 {
+			t.Errorf("(100,5) still present: %v", got)
+		}
+	}
+}
+
+// TestSubscribers_RemoveAllForChat covers the chat-wide prune used when a
+// terminal error means every topic in that chat is dead too.
+func TestSubscribers_RemoveAllForChat(t *testing.T) {
+	ctx := context.Background()
+	kv := storage.NewMemoryKVStore()
+
+	for _, tid := range []int{0, 5, 9} {
+		if _, err := addSubscriber(ctx, kv, 100, tid); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := addSubscriber(ctx, kv, 200, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	n, err := removeAllForChat(ctx, kv, 100)
+	if err != nil {
+		t.Fatalf("removeAllForChat(100): %v", err)
+	}
+	if n != 3 {
+		t.Errorf("removeAllForChat(100) returned %d, want 3", n)
+	}
+
+	got, _ := listSubscribers(ctx, kv)
+	if len(got) != 1 || got[0].ChatID != 200 {
+		t.Errorf("after wipe(100): %v, want only chat 200", got)
+	}
+
+	// Idempotent: second call removes nothing.
+	if n, _ := removeAllForChat(ctx, kv, 100); n != 0 {
+		t.Errorf("second removeAllForChat(100) = %d, want 0", n)
+	}
+}
+
+// TestSubscribers_LegacyDecode locks in backward-compat reads of rows
+// written before topic-aware subscriptions (raw []int64 JSON).
+func TestSubscribers_LegacyDecode(t *testing.T) {
+	ctx := context.Background()
+	kv := storage.NewMemoryKVStore()
+
+	// Write legacy shape directly.
+	if err := kv.Put(ctx, subscribersKey, []byte(`[11,22,33]`)); err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := listSubscribers(ctx, kv)
+	if err != nil {
+		t.Fatalf("listSubscribers: %v", err)
+	}
+	want := []Subscriber{{ChatID: 11}, {ChatID: 22}, {ChatID: 33}}
+	if len(got) != len(want) {
+		t.Fatalf("legacy decode: got %v, want %v", got, want)
+	}
+	for i, s := range got {
+		if s != want[i] {
+			t.Errorf("legacy[%d]: got %v, want %v", i, s, want[i])
+		}
+	}
+
+	// Next mutation rewrites in the new shape — verify by reading raw.
+	if _, err := addSubscriber(ctx, kv, 44, 7); err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := kv.Get(ctx, subscribersKey)
+	if string(raw) == `[11,22,33,44]` {
+		t.Errorf("expected rewrite to new shape, still legacy: %s", raw)
 	}
 }