diff --git a/internal/modules/lolschedule/cron.go b/internal/modules/lolschedule/cron.go index a66461e..c05fdeb 100644 --- a/internal/modules/lolschedule/cron.go +++ b/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 { diff --git a/internal/modules/lolschedule/cron_test.go b/internal/modules/lolschedule/cron_test.go index 526f617..98e0fac 100644 --- a/internal/modules/lolschedule/cron_test.go +++ b/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) } } diff --git a/internal/modules/lolschedule/handlers.go b/internal/modules/lolschedule/handlers.go index 9e9817e..38cea0b 100644 --- a/internal/modules/lolschedule/handlers.go +++ b/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 } diff --git a/internal/modules/lolschedule/handlers_test.go b/internal/modules/lolschedule/handlers_test.go index af52211..9a11c57 100644 --- a/internal/modules/lolschedule/handlers_test.go +++ b/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) } } diff --git a/internal/modules/lolschedule/subscribers.go b/internal/modules/lolschedule/subscribers.go index 5406ddb..1fd35cb 100644 --- a/internal/modules/lolschedule/subscribers.go +++ b/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 +} diff --git a/internal/modules/lolschedule/subscribers_test.go b/internal/modules/lolschedule/subscribers_test.go index 423c129..278fcac 100644 --- a/internal/modules/lolschedule/subscribers_test.go +++ b/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) } }