tiennm99/tiennm99bot
1
0 Fork 0
mirror of https://github.com/tiennm99/tiennm99bot.git synced 2026-10-11 03:13:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(lol): move the subscriber list and daily push fan-out into a shared package

Browse Source 
The subscriber store, the once-per-day claim, the terminal-error
classifier and the throttled fan-out with dead-chat pruning now live in
internal/modules/util/subscription so another module can offer an
opt-in daily push. Stored document shapes and keys are unchanged.
This commit is contained in:
tiennm99 committed 2026-10-02 13:06:07 +07:00
1 parent 53f387c576
commit 59c74de9a9
10 files changed
+488 -458
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/modules/lol/cron.go
+16 -199
View File
@@ -4,79 +4,15 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/go-telegram/bot"
"github.com/go-telegram/bot/models"
"github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/modules"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/modules/util/subscription"
)
// 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 chatWideTerminalMarkers = []string{
"bot was blocked by the user",
"user is deactivated",
"bot is not a member",
"chat not found",
"group chat was upgraded",
"chat was deleted",
}
// 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 terminalNone
}
msg := err.Error()
for _, m := range chatWideTerminalMarkers {
if strings.Contains(msg, m) {
return terminalChatWide
}
}
for _, m := range topicOnlyTerminalMarkers {
if strings.Contains(msg, m) {
return terminalTopicOnly
}
}
return terminalNone
}
// dailyPushCronName is the cron's registry and in-process scheduler key; it
// must be unique across all modules' crons.
const dailyPushCronName = "lol_daily_push"
@@ -92,29 +28,8 @@ const dailyPushSchedule = "0 1 * * *"
// briefly run two containers or operator misconfiguration.
const lastPushDateKey = "daily_push:last_date"
// telegramRateLimitThreshold is the subscriber count above which we throttle
// sends to stay under Telegram's global 30 msg/sec cap. Below it we send hot.
const telegramRateLimitThreshold = 30
// telegramRateLimitDelay is the inter-send pause when above the threshold.
// 50ms = ~20 msg/sec, well clear of the 30/s ceiling with margin for jitter.
const telegramRateLimitDelay = 50 * time.Millisecond
// messageSender is the subset of *bot.Bot the cron handler uses. Defining it
// as an interface lets tests inject a mock without spinning up a fake
// Telegram API server.
type messageSender interface {
SendMessage(ctx context.Context, params *bot.SendMessageParams) (*models.Message, error)
}
// lastPushDoc wraps the last-push date string so it can be stored as a named
// root field in a Mongo document (a bare scalar cannot be a root doc).
type lastPushDoc struct {
Date string `json:"date" bson:"date"`
}
// PushDateStore is the typed store for last-push date documents.
type PushDateStore = storage.DocStore[lastPushDoc]
type PushDateStore = subscription.DayStore
// dailyPushCron returns the cron registration; the in-process scheduler fires
// the handler on Schedule.
@@ -136,44 +51,14 @@ func (s *state) dailyPushHandler(ctx context.Context, deps modules.Deps) error {
return runDailyPush(ctx, s, deps.Bot)
}
// claimDailyPush atomically records that today's ICT schedule-day push is
// happening and reports whether THIS caller won the claim. It is the
// idempotency primitive for the daily push: a winner proceeds to fan out; a
// loser (another trigger already claimed today) returns false and sends
// nothing.
//
// The claim uses version-based optimistic write (PutVersioned) on
// lastPushDateKey so two simultaneous triggers cannot both win.
func claimDailyPush(ctx context.Context, store PushDateStore, pushDay string) (bool, error) {
current, version, err := store.Get(ctx, lastPushDateKey)
switch {
case err == nil:
if current.Date == pushDay {
return false, nil // already pushed today
}
case errors.Is(err, storage.ErrNotFound):
version = 0 // never pushed
default:
return false, err
}
if err := store.PutVersioned(ctx, lastPushDateKey, version, lastPushDoc{Date: pushDay}); err != nil {
if errors.Is(err, storage.ErrConflict) {
return false, nil // another trigger claimed today first
}
return false, err
}
return true, nil
}
// 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.subscribers)
func runDailyPush(ctx context.Context, s *state, sender subscription.Sender) error {
subs, err := subscription.List(ctx, s.subscribers)
if err != nil {
return fmt.Errorf("lol daily push: list subscribers: %w", err)
}
@@ -190,14 +75,13 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
}
filtered := FilterMajor(events)
text := RenderToday(filtered, from)
disableNotification := len(filtered) == 0
// Idempotency gate: claim today's push before sending. A lost claim means
// another trigger already pushed (or is pushing) for this ICT schedule day,
// so we send nothing. Placed after the fetch so a transient fetch failure
// does not consume the day's claim.
pushDay := ictDayKey(from)
won, err := claimDailyPush(ctx, s.pushDate, pushDay)
won, err := subscription.ClaimDay(ctx, s.pushDate, lastPushDateKey, pushDay)
if err != nil {
return fmt.Errorf("lol daily push: claim date: %w", err)
}
@@ -206,86 +90,19 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
return nil
}
throttle := len(subs) > telegramRateLimitThreshold
var sent, failed int
deadChats := map[int64]struct{}{}
var deadTopics []Subscriber
for i, sub := range subs {
if throttle && i > 0 {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(telegramRateLimitDelay):
}
}
if _, err := sender.SendMessage(ctx, &bot.SendMessageParams{
ChatID: sub.ChatID,
MessageThreadID: sub.ThreadID,
Text: text,
ParseMode: models.ParseModeHTML,
DisableNotification: disableNotification,
}); err != nil {
log.Warn("lol daily push send failed",
"chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
failed++
switch classifyTerminal(err) {
case terminalChatWide:
deadChats[sub.ChatID] = struct{}{}
case terminalTopicOnly:
deadTopics = append(deadTopics, sub)
}
continue
}
sent++
res, err := subscription.Fanout(ctx, "lol daily", s.subscribers, &s.subscribersMu, subs, sender, bot.SendMessageParams{
Text: text,
ParseMode: models.ParseModeHTML,
DisableNotification: len(filtered) == 0,
})
if err != nil {
return err
}
// Best-effort prune. A failed write just leaves the dead chats in the list,
// and tomorrow's push fails on them again and retries the prune.
pruned := pruneDeadSubscribers(ctx, s, deadChats, deadTopics)
log.Info("lol daily push complete",
"subscribers", len(subs),
"sent", sent,
"failed", failed,
"pruned", pruned,
"throttled", throttle)
"sent", res.Sent,
"failed", res.Failed,
"pruned", res.Pruned,
"throttled", res.Throttled)
return nil
}
// 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 chatWide {
n, err := removeAllForChat(ctx, s.subscribers, chatID)
if err != nil {
log.Warn("lol 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.subscribers, sub.ChatID, sub.ThreadID)
if err != nil {
log.Warn("lol prune dead topic failed",
"chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
continue
}
if ok {
removed++
}
}
return removed
}
internal/modules/lol/cron_test.go
+23 -59
View File
@@ -13,6 +13,7 @@ import (
"github.com/go-telegram/bot/models"
"github.com/tiennm99/miti99bot/internal/modules"
"github.com/tiennm99/miti99bot/internal/modules/util/subscription"
"github.com/tiennm99/miti99bot/internal/storage"
)
@@ -60,11 +61,11 @@ func fixedNow() time.Time {
// newTestStore builds a fresh set of typed stores over a shared in-memory
// collection, matching what the factory wires in production.
func newTestStore(t *testing.T) (SubscriberStore, PushDateStore, CacheStore) {
func newTestStore(t *testing.T) (subscription.Store, PushDateStore, CacheStore) {
t.Helper()
col := storage.NewMemoryProvider().Collection("lol")
return storage.Typed[subscribersDoc](col),
storage.Typed[lastPushDoc](col),
return storage.Typed[subscription.Doc](col),
storage.Typed[subscription.DayDoc](col),
storage.Typed[cacheRecord](col)
}
@@ -118,7 +119,7 @@ func TestRunDailyPush_SendsEmptyScheduleSilently(t *testing.T) {
chatIDs := []int64{100, 200, 300}
for _, id := range chatIDs {
if _, err := addSubscriber(context.Background(), s.subscribers, id, 0); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, id, 0); err != nil {
t.Fatalf("addSubscriber %d: %v", id, err)
}
}
@@ -166,7 +167,7 @@ func TestRunDailyPush_SendsMatchScheduleWithNotification(t *testing.T) {
},
}})
if _, err := addSubscriber(context.Background(), s.subscribers, 100, 0); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, 100, 0); err != nil {
t.Fatalf("addSubscriber: %v", err)
}
@@ -189,13 +190,13 @@ func TestRunDailyPush_ForwardsMessageThreadID(t *testing.T) {
s := newTestState(t)
seedFreshCache(t, s.cache, nil)
subs := []Subscriber{
subs := []subscription.Subscriber{
{ChatID: 100, ThreadID: 0},
{ChatID: 100, ThreadID: 7},
{ChatID: 200, ThreadID: 42},
}
for _, sub := range subs {
if _, err := addSubscriber(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
t.Fatal(err)
}
}
@@ -226,7 +227,7 @@ func TestRunDailyPush_IdempotentPerICTDay(t *testing.T) {
chatIDs := []int64{100, 200, 300}
for _, id := range chatIDs {
if _, err := addSubscriber(context.Background(), s.subscribers, id, 0); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, id, 0); err != nil {
t.Fatalf("addSubscriber %d: %v", id, err)
}
}
@@ -249,10 +250,10 @@ func TestRunDailyPush_ClaimsICTDayAtMidnight(t *testing.T) {
return time.Date(2026, 5, 9, 17, 0, 0, 0, time.UTC) // 2026-05-10 00:00 ICT
}
seedFreshCache(t, s.cache, nil)
if _, err := addSubscriber(context.Background(), s.subscribers, 100, 0); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, 100, 0); err != nil {
t.Fatal(err)
}
if err := s.pushDate.Put(context.Background(), lastPushDateKey, lastPushDoc{Date: "2026-05-09"}); err != nil {
if err := s.pushDate.Put(context.Background(), lastPushDateKey, subscription.DayDoc{Date: "2026-05-09"}); err != nil {
t.Fatal(err)
}
@@ -278,7 +279,7 @@ func TestRunDailyPush_PartialFailureContinues(t *testing.T) {
chatIDs := []int64{100, 200, 300}
for _, id := range chatIDs {
if _, err := addSubscriber(context.Background(), s.subscribers, id, 0); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, id, 0); err != nil {
t.Fatalf("addSubscriber %d: %v", id, err)
}
}
@@ -302,7 +303,7 @@ func TestRunDailyPush_PrunesDeadSubscribers(t *testing.T) {
// Chat 400 has two topic subs; both should be pruned when the chat
// returns a chat-wide terminal error.
seedSubs := []Subscriber{
seedSubs := []subscription.Subscriber{
{ChatID: 100, ThreadID: 0},
{ChatID: 200, ThreadID: 0},
{ChatID: 300, ThreadID: 0},
@@ -310,7 +311,7 @@ func TestRunDailyPush_PrunesDeadSubscribers(t *testing.T) {
{ChatID: 400, ThreadID: 9},
}
for _, sub := range seedSubs {
if _, err := addSubscriber(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
t.Fatalf("addSubscriber %v: %v", sub, err)
}
}
@@ -323,11 +324,11 @@ func TestRunDailyPush_PrunesDeadSubscribers(t *testing.T) {
t.Fatalf("runDailyPush: %v", err)
}
remaining, err := listSubscribers(context.Background(), s.subscribers)
remaining, err := subscription.List(context.Background(), s.subscribers)
if err != nil {
t.Fatalf("listSubscribers: %v", err)
}
want := []Subscriber{{ChatID: 100}, {ChatID: 200}, {ChatID: 300}}
want := []subscription.Subscriber{{ChatID: 100}, {ChatID: 200}, {ChatID: 300}}
if len(remaining) != len(want) {
t.Fatalf("subscribers after prune: got %v, want %v", remaining, want)
}
@@ -345,13 +346,13 @@ func TestRunDailyPush_TopicOnlyTerminalPrunesOneTopic(t *testing.T) {
s := newTestState(t)
seedFreshCache(t, s.cache, nil)
seedSubs := []Subscriber{
seedSubs := []subscription.Subscriber{
{ChatID: 500, ThreadID: 0},
{ChatID: 500, ThreadID: 11},
{ChatID: 500, ThreadID: 22},
}
for _, sub := range seedSubs {
if _, err := addSubscriber(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
t.Fatal(err)
}
}
@@ -364,7 +365,7 @@ func TestRunDailyPush_TopicOnlyTerminalPrunesOneTopic(t *testing.T) {
t.Fatalf("runDailyPush: %v", err)
}
remaining, _ := listSubscribers(context.Background(), s.subscribers)
remaining, _ := subscription.List(context.Background(), s.subscribers)
if len(remaining) != 0 {
t.Errorf("expected all topic-only entries pruned, got %v", remaining)
}
@@ -378,13 +379,13 @@ func TestRunDailyPush_TopicOnlyTerminalKeepsOtherTopics(t *testing.T) {
// Two distinct chats, each with multiple topic subs. Only chat 600 hits
// the topic-terminal error; chat 700 sends cleanly.
seedSubs := []Subscriber{
seedSubs := []subscription.Subscriber{
{ChatID: 600, ThreadID: 1},
{ChatID: 600, ThreadID: 2},
{ChatID: 700, ThreadID: 3},
}
for _, sub := range seedSubs {
if _, err := addSubscriber(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
if _, err := subscription.Add(context.Background(), s.subscribers, sub.ChatID, sub.ThreadID); err != nil {
t.Fatal(err)
}
}
@@ -395,49 +396,12 @@ func TestRunDailyPush_TopicOnlyTerminalKeepsOtherTopics(t *testing.T) {
t.Fatalf("runDailyPush: %v", err)
}
remaining, _ := listSubscribers(context.Background(), s.subscribers)
if len(remaining) != 1 || remaining[0] != (Subscriber{ChatID: 700, ThreadID: 3}) {
remaining, _ := subscription.List(context.Background(), s.subscribers)
if len(remaining) != 1 || remaining[0] != (subscription.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 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 got := classifyTerminal(errors.New(msg)); got != terminalNone {
t.Errorf("classifyTerminal(%q) = %v, want terminalNone (transient)", msg, got)
}
}
if got := classifyTerminal(nil); got != terminalNone {
t.Errorf("classifyTerminal(nil) = %v, want terminalNone", got)
}
}
func TestDailyPushHandler_NilBot_ReturnsError(t *testing.T) {
s := newTestState(t)
deps := modules.Deps{Store: storage.NewMemoryProvider().Collection("lol")}
internal/modules/lol/handlers.go
+7 -20
View File
@@ -11,11 +11,12 @@ import (
"github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/modules/util/chathelper"
"github.com/tiennm99/miti99bot/internal/modules/util/subscription"
)
// state captures everything a lol handler needs at runtime.
type state struct {
subscribers SubscriberStore
subscribers subscription.Store
pushDate PushDateStore
cache CacheStore
client *Client
@@ -123,20 +124,6 @@ func (s *state) replyForRange(ctx context.Context, b *bot.Bot, msg *models.Messa
return chathelper.ReplyHTML(ctx, b, msg, text)
}
func subscriptionScope(msg *models.Message) string {
if msg != nil && msg.MessageThreadID != 0 {
return "this topic"
}
return "this chat"
}
func subscriptionScopeSentenceSubject(msg *models.Message) string {
if msg != nil && msg.MessageThreadID != 0 {
return "This topic"
}
return "This chat"
}
// handleSubscribe is /lol_subscribe — opt the chat into the daily
// digest delivered by the in-process cron handler.
func (s *state) handleSubscribe(ctx context.Context, b *bot.Bot, update *models.Update) error {
@@ -146,11 +133,11 @@ func (s *state) handleSubscribe(ctx context.Context, b *bot.Bot, update *models.
}
s.subscribersMu.Lock()
defer s.subscribersMu.Unlock()
added, err := addSubscriber(ctx, s.subscribers, msg.Chat.ID, msg.MessageThreadID)
added, err := subscription.Add(ctx, s.subscribers, msg.Chat.ID, msg.MessageThreadID)
if err != nil {
return err
}
scope := subscriptionScope(msg)
scope := subscription.Scope(msg)
if added {
return chathelper.Reply(ctx, b, msg,
"✅ Subscribed "+scope+" to the daily LoL schedule at 08:00 ICT.\n"+
@@ -167,13 +154,13 @@ func (s *state) handleUnsubscribe(ctx context.Context, b *bot.Bot, update *model
}
s.subscribersMu.Lock()
defer s.subscribersMu.Unlock()
removed, err := removeSubscriber(ctx, s.subscribers, msg.Chat.ID, msg.MessageThreadID)
removed, err := subscription.Remove(ctx, s.subscribers, msg.Chat.ID, msg.MessageThreadID)
if err != nil {
return err
}
scope := subscriptionScope(msg)
scope := subscription.Scope(msg)
if removed {
return chathelper.Reply(ctx, b, msg, "Unsubscribed "+scope+".")
}
return chathelper.Reply(ctx, b, msg, subscriptionScopeSentenceSubject(msg)+" wasn't subscribed.")
return chathelper.Reply(ctx, b, msg, subscription.ScopeSubject(msg)+" wasn't subscribed.")
}
internal/modules/lol/handlers_test.go
+14 -13
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/tiennm99/miti99bot/internal/modules"
"github.com/tiennm99/miti99bot/internal/modules/util/subscription"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/testutil"
)
@@ -17,7 +18,7 @@ import (
// custom upstream HTTP server returning bodyJSON for every request. nowMs
// fixes the clock so date-based handlers are deterministic.
// Returns the recording bot and the subscriber store for inspection in tests.
func installSchedule(t *testing.T, bodyJSON string, nowMs int64) (*testutil.RecordingBot, SubscriberStore) {
func installSchedule(t *testing.T, bodyJSON string, nowMs int64) (*testutil.RecordingBot, subscription.Store) {
t.Helper()
setTestToken(t)
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -30,8 +31,8 @@ func installSchedule(t *testing.T, bodyJSON string, nowMs int64) (*testutil.Reco
col := storage.NewMemoryProvider().Collection("lol")
s := &state{
subscribers: storage.Typed[subscribersDoc](col),
pushDate: storage.Typed[lastPushDoc](col),
subscribers: storage.Typed[subscription.Doc](col),
pushDate: storage.Typed[subscription.DayDoc](col),
cache: storage.Typed[cacheRecord](col),
client: &Client{HTTP: upstream.Client(), URL: upstream.URL},
nowFn: func() time.Time { return time.UnixMilli(nowMs).UTC() },
@@ -196,8 +197,8 @@ func TestHandleSubscribe_AddsAndIsIdempotent(t *testing.T) {
if got := rb.LastSent().Text(); !strings.Contains(got, "Already subscribed") {
t.Errorf("duplicate subscribe should report Already; got %q", got)
}
ids, _ := listSubscribers(context.Background(), subsStore)
if len(ids) != 1 || ids[0] != (Subscriber{ChatID: 7}) {
ids, _ := subscription.List(context.Background(), subsStore)
if len(ids) != 1 || ids[0] != (subscription.Subscriber{ChatID: 7}) {
t.Errorf("subscribers = %v, want [{7 0}]", ids)
}
}
@@ -215,8 +216,8 @@ func TestHandleSubscribe_ForumTopic_CapturesThreadID(t *testing.T) {
if got := rb.LastSent().Text(); !strings.Contains(got, "Subscribed this topic") {
t.Errorf("topic subscribe reply = %q", got)
}
subs, _ := listSubscribers(context.Background(), subsStore)
want := Subscriber{ChatID: 555, ThreadID: 42}
subs, _ := subscription.List(context.Background(), subsStore)
want := subscription.Subscriber{ChatID: 555, ThreadID: 42}
if len(subs) != 1 || subs[0] != want {
t.Errorf("subscribers = %v, want [%v]", subs, want)
}
@@ -240,8 +241,8 @@ func TestHandleUnsubscribe_ForumTopic_RemovesOnlyThatTopic(t *testing.T) {
upd.Message.MessageThreadID = 42
rb.Bot.ProcessUpdate(context.Background(), upd)
subs, _ := listSubscribers(context.Background(), subsStore)
want := Subscriber{ChatID: 555, ThreadID: 99}
subs, _ := subscription.List(context.Background(), subsStore)
want := subscription.Subscriber{ChatID: 555, ThreadID: 99}
if len(subs) != 1 || subs[0] != want {
t.Errorf("subscribers = %v, want [%v]", subs, want)
}
@@ -271,8 +272,8 @@ func TestHandleSchedule_UpstreamFailureGivesFriendlyError(t *testing.T) {
rb := testutil.NewRecordingBot(t)
col := storage.NewMemoryProvider().Collection("lol")
s := &state{
subscribers: storage.Typed[subscribersDoc](col),
pushDate: storage.Typed[lastPushDoc](col),
subscribers: storage.Typed[subscription.Doc](col),
pushDate: storage.Typed[subscription.DayDoc](col),
cache: storage.Typed[cacheRecord](col),
client: &Client{HTTP: upstream.Client(), URL: upstream.URL},
nowFn: func() time.Time { return time.UnixMilli(fakeNowMs).UTC() },
@@ -300,8 +301,8 @@ func TestHandleSchedule_CustomDateDoesNotUseFallbackCache(t *testing.T) {
col := storage.NewMemoryProvider().Collection("lol")
cache := storage.Typed[cacheRecord](col)
s := &state{
subscribers: storage.Typed[subscribersDoc](col),
pushDate: storage.Typed[lastPushDoc](col),
subscribers: storage.Typed[subscription.Doc](col),
pushDate: storage.Typed[subscription.DayDoc](col),
cache: cache,
client: &Client{HTTP: upstream.Client(), URL: upstream.URL},
nowFn: func() time.Time { return time.UnixMilli(fakeNowMs).UTC() },
internal/modules/lol/lol.go
+3 -2
View File
@@ -2,6 +2,7 @@ package lol
import (
"github.com/tiennm99/miti99bot/internal/modules"
"github.com/tiennm99/miti99bot/internal/modules/util/subscription"
"github.com/tiennm99/miti99bot/internal/storage"
)
@@ -15,8 +16,8 @@ const CollectionName = "lol"
// without it the handler fails fast with a clear error.
func New(deps modules.Deps) modules.Module {
s := &state{
subscribers: storage.Typed[subscribersDoc](deps.Store),
pushDate: storage.Typed[lastPushDoc](deps.Store),
subscribers: storage.Typed[subscription.Doc](deps.Store),
pushDate: storage.Typed[subscription.DayDoc](deps.Store),
cache: storage.Typed[cacheRecord](deps.Store),
client: &Client{},
}
internal/modules/lol/subscribers.go
-132
View File
@@ -1,132 +0,0 @@
package lol
import (
"context"
"errors"
"fmt"
"github.com/tiennm99/miti99bot/internal/storage"
)
// subscribersKey is the store slot holding the per-module subscriber list.
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" bson:"chat_id"`
ThreadID int `json:"thread_id,omitempty" bson:"thread_id,omitempty"`
}
// subscribersDoc wraps the subscriber list so it can be stored as a named
// root field in a Mongo document (a bare JSON array cannot be a root doc).
type subscribersDoc struct {
Subscribers []Subscriber `json:"subscribers" bson:"subscribers"`
}
// SubscriberStore is the typed store for subscriber documents.
type SubscriberStore = storage.DocStore[subscribersDoc]
// listSubscribers returns the current subscriber list, or an empty slice
// if none have ever subscribed.
func listSubscribers(ctx context.Context, store SubscriberStore) ([]Subscriber, error) {
doc, _, err := store.Get(ctx, subscribersKey)
switch {
case errors.Is(err, storage.ErrNotFound):
return nil, nil
case err != nil:
return nil, fmt.Errorf("lol listSubscribers: %w", err)
}
if doc.Subscribers != nil {
return doc.Subscribers, nil
}
return nil, nil
}
// 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 store 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, store SubscriberStore, chatID int64, threadID int) (bool, error) {
subs, err := listSubscribers(ctx, store)
if err != nil {
return false, err
}
for _, s := range subs {
if s.ChatID == chatID && s.ThreadID == threadID {
return false, nil
}
}
subs = append(subs, Subscriber{ChatID: chatID, ThreadID: threadID})
if err := store.Put(ctx, subscribersKey, subscribersDoc{Subscribers: subs}); err != nil {
return false, fmt.Errorf("lol addSubscriber: %w", err)
}
return true, nil
}
// 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, store SubscriberStore, chatID int64, threadID int) (bool, error) {
subs, err := listSubscribers(ctx, store)
if err != nil {
return false, err
}
out := make([]Subscriber, 0, len(subs))
removed := false
for _, s := range subs {
if s.ChatID == chatID && s.ThreadID == threadID {
removed = true
continue
}
out = append(out, s)
}
if !removed {
return false, nil
}
if err := store.Put(ctx, subscribersKey, subscribersDoc{Subscribers: out}); err != nil {
return false, fmt.Errorf("lol removeSubscriber: %w", err)
}
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, store SubscriberStore, chatID int64) (int, error) {
subs, err := listSubscribers(ctx, store)
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 := store.Put(ctx, subscribersKey, subscribersDoc{Subscribers: out}); err != nil {
return 0, fmt.Errorf("lol removeAllForChat: %w", err)
}
return removed, nil
}
internal/modules/util/subscription/push.go
+211
View File
@@ -0,0 +1,211 @@
package subscription
import (
"context"
"errors"
"strings"
"sync"
"time"
"github.com/go-telegram/bot"
"github.com/go-telegram/bot/models"
"github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/storage"
)
// 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 the next push retries it.
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 a push
// 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 chatWideTerminalMarkers = []string{
"bot was blocked by the user",
"user is deactivated",
"bot is not a member",
"chat not found",
"group chat was upgraded",
"chat was deleted",
}
// 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 TerminalNone
}
msg := err.Error()
for _, m := range chatWideTerminalMarkers {
if strings.Contains(msg, m) {
return TerminalChatWide
}
}
for _, m := range topicOnlyTerminalMarkers {
if strings.Contains(msg, m) {
return TerminalTopicOnly
}
}
return TerminalNone
}
// DayDoc wraps the last-claimed day string so it can be stored as a named
// root field in a Mongo document (a bare scalar cannot be a root doc).
type DayDoc struct {
Date string `json:"date" bson:"date"`
}
// DayStore is the typed store for last-claimed day documents.
type DayStore = storage.DocStore[DayDoc]
// ClaimDay atomically records that the push for day is happening and reports
// whether THIS caller won the claim. A winner proceeds to fan out; a loser
// (another trigger already claimed day) returns false and sends nothing. This
// defends against double-fire windows from rolling deploys that briefly run
// two containers.
//
// The claim uses version-based optimistic write (PutVersioned) on key so two
// simultaneous triggers cannot both win.
func ClaimDay(ctx context.Context, store DayStore, key, day string) (bool, error) {
current, version, err := store.Get(ctx, key)
switch {
case err == nil:
if current.Date == day {
return false, nil
}
case errors.Is(err, storage.ErrNotFound):
version = 0
default:
return false, err
}
if err := store.PutVersioned(ctx, key, version, DayDoc{Date: day}); err != nil {
if errors.Is(err, storage.ErrConflict) {
return false, nil
}
return false, err
}
return true, nil
}
// rateLimitThreshold is the subscriber count above which sends are throttled
// to stay under Telegram's global 30 msg/sec cap. Below it we send hot.
const rateLimitThreshold = 30
// rateLimitDelay is the inter-send pause when above the threshold. 50ms =
// ~20 msg/sec, well clear of the 30/s ceiling with margin for jitter.
const rateLimitDelay = 50 * time.Millisecond
// Sender is the subset of *bot.Bot a push uses. Defining it as an interface
// lets tests inject a mock without spinning up a fake Telegram API server.
type Sender interface {
SendMessage(ctx context.Context, params *bot.SendMessageParams) (*models.Message, error)
}
// Result summarizes one fan-out.
type Result struct {
Sent, Failed, Pruned int
Throttled bool
}
// Fanout sends params(sub) to every subscriber. Per-chat send failures are
// logged but do not abort the batch — one bad chat does not deny the rest.
// ChatID and MessageThreadID are filled in from each subscriber so forum
// subscribers receive the push in their topic, not in General.
//
// Unreachable subscribers are pruned afterwards, best-effort, while holding
// mu (the same mutex that guards Add and Remove). A failed prune just leaves
// the dead chats listed, and the next push fails on them and retries.
// The only returned error is ctx cancellation while throttling.
func Fanout(ctx context.Context, name string, store Store, mu *sync.Mutex, subs []Subscriber, sender Sender, params bot.SendMessageParams) (Result, error) {
res := Result{Throttled: len(subs) > rateLimitThreshold}
deadChats := map[int64]struct{}{}
var deadTopics []Subscriber
for i, sub := range subs {
if res.Throttled && i > 0 {
select {
case <-ctx.Done():
return res, ctx.Err()
case <-time.After(rateLimitDelay):
}
}
p := params
p.ChatID = sub.ChatID
p.MessageThreadID = sub.ThreadID
if _, err := sender.SendMessage(ctx, &p); err != nil {
log.Warn(name+" push send failed", "chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
res.Failed++
switch ClassifyTerminal(err) {
case TerminalChatWide:
deadChats[sub.ChatID] = struct{}{}
case TerminalTopicOnly:
deadTopics = append(deadTopics, sub)
}
continue
}
res.Sent++
}
res.Pruned = prune(ctx, name, store, mu, deadChats, deadTopics)
return res, nil
}
// prune removes entries flagged unreachable. Chat-wide failures drop every
// subscription for the chat; topic-only failures drop just the one
// (ChatID, ThreadID). Returns total entries removed.
func prune(ctx context.Context, name string, store Store, mu *sync.Mutex, chatWide map[int64]struct{}, topicOnly []Subscriber) int {
if len(chatWide) == 0 && len(topicOnly) == 0 {
return 0
}
mu.Lock()
defer mu.Unlock()
removed := 0
for chatID := range chatWide {
n, err := RemoveAllForChat(ctx, store, chatID)
if err != nil {
log.Warn(name+" prune dead chat failed", "chat", chatID, "err", err)
continue
}
removed += n
}
for _, sub := range topicOnly {
// The whole chat was already pruned above.
if _, ok := chatWide[sub.ChatID]; ok {
continue
}
ok, err := Remove(ctx, store, sub.ChatID, sub.ThreadID)
if err != nil {
log.Warn(name+" prune dead topic failed", "chat", sub.ChatID, "thread", sub.ThreadID, "err", err)
continue
}
if ok {
removed++
}
}
return removed
}
internal/modules/util/subscription/push_test.go
+43
View File
@@ -0,0 +1,43 @@
package subscription
import (
"errors"
"testing"
)
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 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 got := ClassifyTerminal(errors.New(msg)); got != TerminalNone {
t.Errorf("ClassifyTerminal(%q) = %v, want TerminalNone (transient)", msg, got)
}
}
if got := ClassifyTerminal(nil); got != TerminalNone {
t.Errorf("ClassifyTerminal(nil) = %v, want TerminalNone", got)
}
}
internal/modules/util/subscription/subscription.go
+138
View File
@@ -0,0 +1,138 @@
// Package subscription holds the opt-in list behind a module's daily push:
// the stored (chat, topic) subscribers, the once-per-day claim that keeps a
// push idempotent, and the fan-out that sends to every subscriber and prunes
// chats that can no longer be reached.
package subscription
import (
"context"
"errors"
"fmt"
"github.com/go-telegram/bot/models"
"github.com/tiennm99/miti99bot/internal/storage"
)
// Key is the store slot holding a module's subscriber list.
const Key = "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" bson:"chat_id"`
ThreadID int `json:"thread_id,omitempty" bson:"thread_id,omitempty"`
}
// Doc wraps the subscriber list so it can be stored as a named root field in
// a Mongo document (a bare JSON array cannot be a root doc).
type Doc struct {
Subscribers []Subscriber `json:"subscribers" bson:"subscribers"`
}
// Store is the typed store for subscriber documents.
type Store = storage.DocStore[Doc]
// List returns the current subscriber list, or nil if none have ever
// subscribed.
func List(ctx context.Context, store Store) ([]Subscriber, error) {
doc, _, err := store.Get(ctx, Key)
switch {
case errors.Is(err, storage.ErrNotFound):
return nil, nil
case err != nil:
return nil, fmt.Errorf("subscription list: %w", err)
}
return doc.Subscribers, nil
}
// Add 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 store slot, so a concurrent
// Get→mutate→Put from two chats subscribing in the same millisecond would
// drop one write. Callers MUST serialize through a module-scoped mutex (the
// same one passed to Fanout) before calling this.
func Add(ctx context.Context, store Store, chatID int64, threadID int) (bool, error) {
subs, err := List(ctx, store)
if err != nil {
return false, err
}
for _, s := range subs {
if s.ChatID == chatID && s.ThreadID == threadID {
return false, nil
}
}
subs = append(subs, Subscriber{ChatID: chatID, ThreadID: threadID})
if err := store.Put(ctx, Key, Doc{Subscribers: subs}); err != nil {
return false, fmt.Errorf("subscription add: %w", err)
}
return true, nil
}
// Remove 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 Add; callers must hold the
// module's subscriber mutex.
func Remove(ctx context.Context, store Store, chatID int64, threadID int) (bool, error) {
n, err := removeWhere(ctx, store, func(s Subscriber) bool {
return s.ChatID == chatID && s.ThreadID == threadID
})
return n > 0, err
}
// 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 the module's subscriber mutex.
func RemoveAllForChat(ctx context.Context, store Store, chatID int64) (int, error) {
return removeWhere(ctx, store, func(s Subscriber) bool { return s.ChatID == chatID })
}
func removeWhere(ctx context.Context, store Store, drop func(Subscriber) bool) (int, error) {
subs, err := List(ctx, store)
if err != nil {
return 0, err
}
out := make([]Subscriber, 0, len(subs))
for _, s := range subs {
if !drop(s) {
out = append(out, s)
}
}
removed := len(subs) - len(out)
if removed == 0 {
return 0, nil
}
if err := store.Put(ctx, Key, Doc{Subscribers: out}); err != nil {
return 0, fmt.Errorf("subscription remove: %w", err)
}
return removed, nil
}
// Scope names where a subscription applies, for use mid-sentence:
// "this topic" inside a forum topic, otherwise "this chat".
func Scope(msg *models.Message) string {
if msg != nil && msg.MessageThreadID != 0 {
return "this topic"
}
return "this chat"
}
// ScopeSubject is Scope capitalized for the start of a sentence.
func ScopeSubject(msg *models.Message) string {
if msg != nil && msg.MessageThreadID != 0 {
return "This topic"
}
return "This chat"
}
internal/modules/lol/subscribers_test.go → internal/modules/util/subscription/subscription_test.go
+33 -33
View File
@@ -1,4 +1,4 @@
package lol
package subscription
import (
"context"
@@ -7,51 +7,51 @@ import (
"github.com/tiennm99/miti99bot/internal/storage"
)
func newSubscriberStore() SubscriberStore {
return storage.Typed[subscribersDoc](storage.NewMemoryProvider().Collection("lol"))
func newStore() Store {
return storage.Typed[Doc](storage.NewMemoryProvider().Collection("test"))
}
func TestSubscribers_AddRemoveListIdempotent(t *testing.T) {
ctx := context.Background()
store := newSubscriberStore()
store := newStore()
got, _ := listSubscribers(ctx, store)
got, _ := List(ctx, store)
if len(got) != 0 {
t.Errorf("empty list = %v, want []", got)
}
added, err := addSubscriber(ctx, store, 42, 0)
added, err := Add(ctx, store, 42, 0)
if err != nil || !added {
t.Fatalf("first add: added=%v err=%v", added, err)
}
added, _ = addSubscriber(ctx, store, 42, 0)
added, _ = Add(ctx, store, 42, 0)
if added {
t.Errorf("re-add should be no-op")
}
got, _ = listSubscribers(ctx, store)
got, _ = List(ctx, store)
if len(got) != 1 || got[0] != (Subscriber{ChatID: 42}) {
t.Errorf("after add(42,0): %v, want [{42 0}]", got)
}
if _, err := addSubscriber(ctx, store, 7, 0); err != nil {
if _, err := Add(ctx, store, 7, 0); err != nil {
t.Fatal(err)
}
got, _ = listSubscribers(ctx, store)
got, _ = List(ctx, store)
if len(got) != 2 {
t.Errorf("after add(7,0): %v, want len 2", got)
}
removed, err := removeSubscriber(ctx, store, 42, 0)
removed, err := Remove(ctx, store, 42, 0)
if err != nil || !removed {
t.Fatalf("remove(42,0): removed=%v err=%v", removed, err)
}
removed, _ = removeSubscriber(ctx, store, 42, 0)
removed, _ = Remove(ctx, store, 42, 0)
if removed {
t.Errorf("re-remove should be no-op")
}
got, _ = listSubscribers(ctx, store)
got, _ = List(ctx, store)
if len(got) != 1 || got[0].ChatID != 7 {
t.Errorf("after remove(42,0): %v, want [{7 0}]", got)
}
@@ -62,29 +62,29 @@ func TestSubscribers_AddRemoveListIdempotent(t *testing.T) {
// topic must not affect the others.
func TestSubscribers_TopicsAreDistinct(t *testing.T) {
ctx := context.Background()
store := newSubscriberStore()
store := newStore()
for _, tid := range []int{0, 5, 9} {
added, err := addSubscriber(ctx, store, 100, tid)
added, err := Add(ctx, store, 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, store, 100, 5); added {
if added, _ := Add(ctx, store, 100, 5); added {
t.Errorf("duplicate (100,5) should be no-op")
}
got, _ := listSubscribers(ctx, store)
got, _ := List(ctx, store)
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, store, 100, 5); !removed {
if removed, _ := Remove(ctx, store, 100, 5); !removed {
t.Error("remove(100,5) should report removed=true")
}
got, _ = listSubscribers(ctx, store)
got, _ = List(ctx, store)
if len(got) != 2 {
t.Errorf("after remove(100,5): %v, want len 2", got)
}
@@ -99,47 +99,47 @@ func TestSubscribers_TopicsAreDistinct(t *testing.T) {
// terminal error means every topic in that chat is dead too.
func TestSubscribers_RemoveAllForChat(t *testing.T) {
ctx := context.Background()
store := newSubscriberStore()
store := newStore()
for _, tid := range []int{0, 5, 9} {
if _, err := addSubscriber(ctx, store, 100, tid); err != nil {
if _, err := Add(ctx, store, 100, tid); err != nil {
t.Fatal(err)
}
}
if _, err := addSubscriber(ctx, store, 200, 0); err != nil {
if _, err := Add(ctx, store, 200, 0); err != nil {
t.Fatal(err)
}
n, err := removeAllForChat(ctx, store, 100)
n, err := RemoveAllForChat(ctx, store, 100)
if err != nil {
t.Fatalf("removeAllForChat(100): %v", err)
t.Fatalf("RemoveAllForChat(100): %v", err)
}
if n != 3 {
t.Errorf("removeAllForChat(100) returned %d, want 3", n)
t.Errorf("RemoveAllForChat(100) returned %d, want 3", n)
}
got, _ := listSubscribers(ctx, store)
got, _ := List(ctx, store)
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, store, 100); n != 0 {
t.Errorf("second removeAllForChat(100) = %d, want 0", n)
if n, _ := RemoveAllForChat(ctx, store, 100); n != 0 {
t.Errorf("second RemoveAllForChat(100) = %d, want 0", n)
}
}
func TestSubscribers_CurrentShapeRoundTrip(t *testing.T) {
ctx := context.Background()
store := newSubscriberStore()
store := newStore()
// Write the current shape directly via Put so we can confirm round-trip.
currentSubs := []Subscriber{{ChatID: 11}, {ChatID: 22}, {ChatID: 33}}
if err := store.Put(ctx, subscribersKey, subscribersDoc{Subscribers: currentSubs}); err != nil {
if err := store.Put(ctx, Key, Doc{Subscribers: currentSubs}); err != nil {
t.Fatal(err)
}
got, err := listSubscribers(ctx, store)
got, err := List(ctx, store)
if err != nil {
t.Fatalf("listSubscribers: %v", err)
}
@@ -154,10 +154,10 @@ func TestSubscribers_CurrentShapeRoundTrip(t *testing.T) {
}
// Next mutation verifies subscribers wrap correctly.
if _, err := addSubscriber(ctx, store, 44, 7); err != nil {
if _, err := Add(ctx, store, 44, 7); err != nil {
t.Fatal(err)
}
doc, _, _ := store.Get(ctx, subscribersKey)
doc, _, _ := store.Get(ctx, Key)
if doc.Subscribers == nil {
t.Error("expected non-nil Subscribers field after add")
}