From 6fd65729ac56063c1098dfde687ca9befcbff6b6 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Wed, 1 Jul 2026 10:22:03 +0700 Subject: [PATCH] feat(stats): use queryable Mongo usage records --- Makefile | 2 +- README.md | 5 +- cmd/server/main.go | 6 + docs/deploy-coolify-selfhosted.md | 10 +- internal/modules/stats/startup.go | 215 +++++++++ internal/modules/stats/startup_mongo_test.go | 100 ++++ internal/modules/stats/stats.go | 131 +---- internal/modules/stats/stats_test.go | 185 +++++-- internal/modules/stats/usage_store.go | 451 ++++++++++++++++++ internal/modules/stats/views.go | 197 +------- internal/storage/mongodb_provider.go | 11 + internal/systemstate/systemstate.go | 46 ++ ...0701-1000-mongodb-stats-design-research.md | 265 ++++++++++ 13 files changed, 1275 insertions(+), 349 deletions(-) create mode 100644 internal/modules/stats/startup.go create mode 100644 internal/modules/stats/startup_mongo_test.go create mode 100644 internal/modules/stats/usage_store.go create mode 100644 internal/systemstate/systemstate.go create mode 100644 plans/reports/260701-1000-mongodb-stats-design-research.md diff --git a/Makefile b/Makefile index 26e4f61..1cb322a 100644 --- a/Makefile +++ b/Makefile @@ -20,7 +20,7 @@ test: ## Unit tests (no emulator required) MONGO_PORT ?= 27017 test-mongo: mongo-local ## Run MongoDB tests against a local Mongo container MONGODB_TEST_URL=mongodb://127.0.0.1:$(MONGO_PORT) LOG_LEVEL=error \ - go test -race -count=1 ./internal/storage/... + go test -race -count=1 ./internal/storage/... ./internal/modules/stats/... # ---- Lint / Vet ------------------------------------------------------------ diff --git a/README.md b/README.md index 67a87c6..33cb2cb 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ internal/telegram/ Telegram long-polling bot wrapper internal/cron/ in-process cron scheduler internal/modules/ Module framework, registry, dispatchers, modules internal/storage/ typed DocStore[T] (Provider + Typed); mongodb runtime + memory (tests). Values persist as flattened native BSON root documents +internal/systemstate/ shared `system` collection metadata for startup migrations compose.yml Coolify self-host stack (single bot service) telegram-commands.json Manual Telegram command menu source docs/deploy-coolify-selfhosted.md Self-host deploy and operations guide @@ -61,7 +62,7 @@ go run ./cmd/server For integration tests (each skips when its emulator env var is unset): ```sh make mongo-local # docker run mongo:7 on :27017 -make test-mongo # internal/storage typed-store tests against local MongoDB +make test-mongo # MongoDB integration tests against local MongoDB ``` ## Test @@ -69,7 +70,7 @@ make test-mongo # internal/storage typed-store tests against local Mong ```sh make vet # go vet make test # full unit suite (no emulator) -make test-mongo # typed-store integration tests against local Mongo (requires Docker) +make test-mongo # MongoDB integration tests against local Mongo (requires Docker) ``` ## Deploy diff --git a/cmd/server/main.go b/cmd/server/main.go index 7f7ef2a..f70e679 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -29,6 +29,7 @@ import ( "github.com/tiennm99/miti99bot/internal/modules/wordle" "github.com/tiennm99/miti99bot/internal/server" "github.com/tiennm99/miti99bot/internal/storage" + "github.com/tiennm99/miti99bot/internal/systemstate" "github.com/tiennm99/miti99bot/internal/telegram" ) @@ -93,6 +94,11 @@ func main() { log.Fatal("storage init failed", "err", err) } defer closeProvider() + systemColl := provider.Collection(systemstate.CollectionName) + + if err := stats.InitStore(rootCtx, provider.Collection("stats"), systemColl); err != nil { + log.Fatal("stats storage init failed", "err", err) + } b, err := telegram.NewBot(cfg.TelegramBotToken) if err != nil { diff --git a/docs/deploy-coolify-selfhosted.md b/docs/deploy-coolify-selfhosted.md index d7096b4..aed59e1 100644 --- a/docs/deploy-coolify-selfhosted.md +++ b/docs/deploy-coolify-selfhosted.md @@ -8,7 +8,7 @@ Run `miti99bot` as a long-lived container on [Coolify](https://coolify.io) with ``` Telegram <── long poll (getUpdates) ── container (outbound only) in-process scheduler ───────────────────> module crons - MongoDB Atlas (db / one collection per module) + MongoDB Atlas (db / one collection per module + system metadata) Coolify env vars (plain secrets) NO public ingress (polling = outbound only; no domain, no /webhook, no TLS in) ``` @@ -60,13 +60,19 @@ overrides are not supported in runtime env; modules use coded defaults. 4. Copy the `mongodb+srv://…` connection string into `MONGO_URL` and put the db name in `MONGO_DATABASE`. -> Storage layout: one collection per module; each document is a flattened native +> Storage layout: one collection per module plus a shared `system` collection +> for startup metadata such as one-time migrations. Each document is a flattened native > document — `{ _id: , ...payload fields, version, updatedAt }` with no > `value` envelope. Payload fields are hoisted to the document root so they > expand and are queryable in Compass. The two non-object values are wrapped in a > named field: schedule subscribers under `subscribers` (array) and the daily > push date under `date`. Concurrency uses the `version` field (optimistic lock); > `updatedAt` is a BSON Date. +> +> The `stats` collection uses queryable aggregate documents for command/user +> counts and creates indexes on startup. First startup after the schema change +> migrates legacy `count:`, `user:`, and `pair:` stats keys into the new shape, +> deletes the legacy keys, and records completion in `system`. ## 2. Coolify diff --git a/internal/modules/stats/startup.go b/internal/modules/stats/startup.go new file mode 100644 index 0000000..80daf26 --- /dev/null +++ b/internal/modules/stats/startup.go @@ -0,0 +1,215 @@ +package stats + +import ( + "context" + "fmt" + "strconv" + "strings" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + + "github.com/tiennm99/miti99bot/internal/storage" + "github.com/tiennm99/miti99bot/internal/systemstate" +) + +const ( + legacyCountPrefix = "count:" + legacyUserPrefix = "user:" + legacyPairPrefix = "pair:" + + usageMigrationName = "stats-usage-v2" + usageMigrationKey = "migration:" + usageMigrationName +) + +type legacyCountEntry struct { + N int64 `json:"n" bson:"n"` +} + +type legacyUserEntry struct { + Username string `json:"username" bson:"username"` + N int64 `json:"n" bson:"n"` +} + +// InitStore performs stats collection startup maintenance. It is safe to call +// every boot: indexes are idempotent and legacy migration is guarded by a +// system collection marker. +func InitStore(ctx context.Context, statsColl, systemColl storage.Collection) error { + if mongoColl, ok := storage.MongoCollection(statsColl); ok { + if err := ensureUsageIndexes(ctx, mongoColl); err != nil { + return err + } + } + return migrateLegacyUsage(ctx, statsColl, systemColl) +} + +func ensureUsageIndexes(ctx context.Context, coll *mongo.Collection) error { + models := []mongo.IndexModel{ + { + Keys: bson.D{bsonField("cmd", 1), bsonField("n", -1), bsonField("user", 1)}, + Options: options.Index().SetName("stats_cmd_n_user"), + }, + { + Keys: bson.D{bsonField("uid", 1), bsonField("n", -1), bsonField("cmd", 1)}, + Options: options.Index().SetName("stats_uid_n_cmd"), + }, + { + Keys: bson.D{bsonField("user", 1), bsonField("updatedAt", -1)}, + Options: options.Index().SetName("stats_user_updated_at"), + }, + } + if _, err := coll.Indexes().CreateMany(ctx, models); err != nil { + return fmt.Errorf("stats indexes: %w", err) + } + return nil +} + +func migrateLegacyUsage(ctx context.Context, statsColl, systemColl storage.Collection) error { + sys := systemstate.New(systemColl) + if rec, ok, err := sys.Get(ctx, usageMigrationKey); err != nil { + return fmt.Errorf("stats legacy migration marker get: %w", err) + } else if ok && rec.Status == "done" { + return nil + } + + legacyCounts := storage.Typed[legacyCountEntry](statsColl) + legacyUsers := storage.Typed[legacyUserEntry](statsColl) + usageDocs := storage.Typed[usageEntry](statsColl) + + usernames, userKeys, err := loadLegacyUsernames(ctx, legacyUsers) + if err != nil { + return err + } + countTotals, countKeys, err := loadLegacyCounts(ctx, legacyCounts) + if err != nil { + return err + } + pairs, pairKeys, pairSums, err := loadLegacyPairs(ctx, legacyCounts, usernames) + if err != nil { + return err + } + + written := int64(0) + for _, entry := range pairs { + if err := usageDocs.Put(ctx, usageKey(entry.Cmd, entry.UserID), entry); err != nil { + return fmt.Errorf("stats legacy pair put %s:%d: %w", entry.Cmd, entry.UserID, err) + } + written++ + } + + for cmd, total := range countTotals { + anonymous := total - pairSums[cmd] + if anonymous <= 0 { + continue + } + if err := usageDocs.Put(ctx, usageKey(cmd, 0), usageEntry{Cmd: cmd, N: anonymous}); err != nil { + return fmt.Errorf("stats legacy anonymous put %s: %w", cmd, err) + } + written++ + } + + for _, key := range append(append(countKeys, pairKeys...), userKeys...) { + if err := legacyCounts.Delete(ctx, key); err != nil { + return fmt.Errorf("stats legacy delete %s: %w", key, err) + } + } + + now := nowMillis() + if err := sys.Put(ctx, usageMigrationKey, systemstate.Record{ + Kind: "migration", + Name: usageMigrationName, + Status: "done", + Count: written, + CompletedAt: now, + UpdatedAt: now, + }); err != nil { + return fmt.Errorf("stats legacy migration marker put: %w", err) + } + return nil +} + +func loadLegacyUsernames(ctx context.Context, users storage.DocStore[legacyUserEntry]) (map[int64]string, []string, error) { + keys, err := users.List(ctx, legacyUserPrefix) + if err != nil { + return nil, nil, fmt.Errorf("stats legacy users list: %w", err) + } + usernames := make(map[int64]string, len(keys)) + for _, key := range keys { + id, err := strconv.ParseInt(strings.TrimPrefix(key, legacyUserPrefix), 10, 64) + if err != nil { + continue + } + entry, _, err := users.Get(ctx, key) + if err != nil { + return nil, nil, fmt.Errorf("stats legacy user get %s: %w", key, err) + } + usernames[id] = entry.Username + } + return usernames, keys, nil +} + +func loadLegacyCounts(ctx context.Context, counts storage.DocStore[legacyCountEntry]) (map[string]int64, []string, error) { + keys, err := counts.List(ctx, legacyCountPrefix) + if err != nil { + return nil, nil, fmt.Errorf("stats legacy counts list: %w", err) + } + totals := make(map[string]int64, len(keys)) + for _, key := range keys { + cmd := strings.TrimPrefix(key, legacyCountPrefix) + if cmd == "" { + continue + } + entry, _, err := counts.Get(ctx, key) + if err != nil { + return nil, nil, fmt.Errorf("stats legacy count get %s: %w", key, err) + } + totals[cmd] = entry.N + } + return totals, keys, nil +} + +func loadLegacyPairs(ctx context.Context, counts storage.DocStore[legacyCountEntry], usernames map[int64]string) ([]usageEntry, []string, map[string]int64, error) { + keys, err := counts.List(ctx, legacyPairPrefix) + if err != nil { + return nil, nil, nil, fmt.Errorf("stats legacy pairs list: %w", err) + } + pairs := make([]usageEntry, 0, len(keys)) + pairSums := make(map[string]int64) + for _, key := range keys { + cmd, userID, ok := parseLegacyPairKey(key) + if !ok { + continue + } + entry, _, err := counts.Get(ctx, key) + if err != nil { + return nil, nil, nil, fmt.Errorf("stats legacy pair get %s: %w", key, err) + } + pairs = append(pairs, usageEntry{ + Cmd: cmd, + UserID: userID, + Username: usernames[userID], + N: entry.N, + }) + pairSums[cmd] += entry.N + } + return pairs, keys, pairSums, nil +} + +func parseLegacyPairKey(key string) (string, int64, bool) { + rest := strings.TrimPrefix(key, legacyPairPrefix) + idx := strings.LastIndexByte(rest, ':') + if idx <= 0 || idx == len(rest)-1 { + return "", 0, false + } + id, err := strconv.ParseInt(rest[idx+1:], 10, 64) + if err != nil { + return "", 0, false + } + return rest[:idx], id, true +} + +func nowMillis() int64 { + return time.Now().UnixMilli() +} diff --git a/internal/modules/stats/startup_mongo_test.go b/internal/modules/stats/startup_mongo_test.go new file mode 100644 index 0000000..f837362 --- /dev/null +++ b/internal/modules/stats/startup_mongo_test.go @@ -0,0 +1,100 @@ +package stats + +import ( + "context" + "errors" + "fmt" + "os" + "testing" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + + "github.com/tiennm99/miti99bot/internal/storage" + "github.com/tiennm99/miti99bot/internal/systemstate" +) + +func TestInitStore_MongoCreatesIndexesAndMigratesLegacy(t *testing.T) { + uri := os.Getenv("MONGODB_TEST_URL") + if uri == "" { + t.Skip("MONGODB_TEST_URL not set; skipping MongoDB integration test") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + client, err := storage.NewMongoClient(ctx, uri) + if err != nil { + t.Fatalf("NewMongoClient: %v", err) + } + dbName := fmt.Sprintf("miti99bot_stats_test_%d", time.Now().UnixNano()) + db := client.Database(dbName) + defer func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = db.Drop(cleanupCtx) + _ = client.Disconnect(cleanupCtx) + }() + + provider := storage.NewMongoProvider(db) + statsColl := provider.Collection("stats") + systemColl := provider.Collection(systemstate.CollectionName) + legacyCounts := storage.Typed[legacyCountEntry](statsColl) + legacyUsers := storage.Typed[legacyUserEntry](statsColl) + if err := legacyCounts.Put(ctx, legacyCountPrefix+"ping", legacyCountEntry{N: 2}); err != nil { + t.Fatalf("legacy count: %v", err) + } + if err := legacyUsers.Put(ctx, legacyUserPrefix+"7", legacyUserEntry{Username: "alice", N: 1}); err != nil { + t.Fatalf("legacy user: %v", err) + } + if err := legacyCounts.Put(ctx, legacyPairPrefix+"ping:7", legacyCountEntry{N: 1}); err != nil { + t.Fatalf("legacy pair: %v", err) + } + + if err := InitStore(ctx, statsColl, systemColl); err != nil { + t.Fatalf("InitStore: %v", err) + } + + usageDocs := storage.Typed[usageEntry](statsColl) + assertUsageEntry(t, usageDocs, usageKey("ping", 7), usageEntry{Cmd: "ping", UserID: 7, Username: "alice", N: 1}) + assertUsageEntry(t, usageDocs, usageKey("ping", 0), usageEntry{Cmd: "ping", N: 1}) + + rawStatsColl, ok := storage.MongoCollection(statsColl) + if !ok { + t.Fatal("stats collection is not Mongo-backed") + } + cur, err := rawStatsColl.Indexes().List(ctx) + if err != nil { + t.Fatalf("list indexes: %v", err) + } + defer func() { _ = cur.Close(ctx) }() + + found := map[string]bool{} + for cur.Next(ctx) { + var doc struct { + Name string `bson:"name"` + } + if err := cur.Decode(&doc); err != nil { + t.Fatalf("decode index: %v", err) + } + found[doc.Name] = true + } + if err := cur.Err(); err != nil { + t.Fatalf("index cursor: %v", err) + } + for _, name := range []string{"stats_cmd_n_user", "stats_uid_n_cmd", "stats_user_updated_at"} { + if !found[name] { + t.Fatalf("missing index %s; indexes=%v", name, found) + } + } + + rawDoc := bson.M{} + err = rawStatsColl.FindOne(ctx, bson.M{"_id": legacyCountPrefix + "ping"}).Decode(&rawDoc) + if err == nil { + t.Fatalf("legacy count key still exists: %+v", rawDoc) + } + if !errors.Is(err, mongo.ErrNoDocuments) { + t.Fatalf("legacy count lookup err = %v, want ErrNoDocuments", err) + } +} diff --git a/internal/modules/stats/stats.go b/internal/modules/stats/stats.go index 9a70dd0..b5ff6cc 100644 --- a/internal/modules/stats/stats.go +++ b/internal/modules/stats/stats.go @@ -1,12 +1,9 @@ -// Package stats tracks per-command and per-user invocation counts persistently -// in DocStore and exposes /stats subcommands to display them sorted by popularity. +// Package stats tracks command usage and exposes /stats subcommands sorted by +// popularity. package stats import ( "context" - "errors" - "strconv" - "sync" "github.com/go-telegram/bot/models" @@ -15,128 +12,46 @@ import ( "github.com/tiennm99/miti99bot/internal/storage" ) -// Sort-key shapes inside the stats module's partition: -// -// count: → countEntry — total per command -// user: → userEntry — cached username + total per user -// pair:: → countEntry — per (command, user) pair -const ( - countPrefix = "count:" - userPrefix = "user:" - pairPrefix = "pair:" - topK = 20 -) +const topK = 20 -type countEntry struct { - N int64 `json:"n" bson:"n"` -} - -type userEntry struct { - Username string `json:"username" bson:"username"` - N int64 `json:"n" bson:"n"` -} - -// counter holds two typed views over the same module Collection: -// one for countEntry values and one for userEntry values. -// Keys are disjoint by prefix so both views safely share the collection. +// counter owns the stats repository used by the command hook and render views. type counter struct { - counts storage.DocStore[countEntry] - users storage.DocStore[userEntry] + store usageStore } -func countKey(name string) string { return countPrefix + name } - -func userKey(id int64) string { return userPrefix + strconv.FormatInt(id, 10) } - -func pairKey(cmd string, id int64) string { - return pairPrefix + cmd + ":" + strconv.FormatInt(id, 10) -} - -// Inc fans out persistent counter writes for one command invocation. -// Always increments count:. When the originating user has a Telegram -// username, also increments user: (refreshing the cached username) and -// pair::. Errors are logged and swallowed; concurrent invocations of -// the same (cmd, user) may lose updates — stats are best-effort. A future -// backend atomic increment would close the race. +// Inc records one authorized command invocation. A sender only contributes to +// user-level stats when Telegram provides a public username; otherwise the +// invocation still contributes to command totals. func (c *counter) Inc(ctx context.Context, name string, update *models.Update) { var ( - userID int64 - username string - hasUser bool + user usageUser + hasUser bool ) if update != nil && update.Message != nil && update.Message.From != nil && update.Message.From.Username != "" { - userID = update.Message.From.ID - username = update.Message.From.Username + user = usageUser{ + ID: update.Message.From.ID, + Username: update.Message.From.Username, + } hasUser = true } - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - c.incCount(ctx, name) - }() - if hasUser { - wg.Add(2) - go func() { - defer wg.Done() - c.incUser(ctx, userID, username) - }() - go func() { - defer wg.Done() - c.incPair(ctx, name, userID) - }() - } - wg.Wait() -} - -func (c *counter) incCount(ctx context.Context, name string) { - key := countKey(name) - entry, _, err := c.counts.Get(ctx, key) - if err != nil && !errors.Is(err, storage.ErrNotFound) { - log.Error("stats: store get failed", "key", key, "err", err) - return - } - entry.N++ - if err := c.counts.Put(ctx, key, entry); err != nil { - log.Error("stats: store put failed", "key", key, "err", err) + if err := c.store.Increment(ctx, name, user, hasUser); err != nil { + if hasUser { + log.Error("stats: increment failed", "command", name, "user_id", user.ID, "err", err) + return + } + log.Error("stats: increment failed", "command", name, "err", err) } } -func (c *counter) incUser(ctx context.Context, id int64, username string) { - key := userKey(id) - entry, _, err := c.users.Get(ctx, key) - if err != nil && !errors.Is(err, storage.ErrNotFound) { - log.Error("stats: store get failed", "key", key, "err", err) - return - } - entry.Username = username - entry.N++ - if err := c.users.Put(ctx, key, entry); err != nil { - log.Error("stats: store put failed", "key", key, "err", err) - } -} - -func (c *counter) incPair(ctx context.Context, name string, id int64) { - key := pairKey(name, id) - entry, _, err := c.counts.Get(ctx, key) - if err != nil && !errors.Is(err, storage.ErrNotFound) { - log.Error("stats: store get failed", "key", key, "err", err) - return - } - entry.N++ - if err := c.counts.Put(ctx, key, entry); err != nil { - log.Error("stats: store put failed", "key", key, "err", err) - } +func newCounter(coll storage.Collection) *counter { + return &counter{store: newUsageStore(coll)} } // New is the module Factory. Registers a CommandHook that persists counts and // a /stats command that displays them. func New(deps modules.Deps) modules.Module { - c := &counter{ - counts: storage.Typed[countEntry](deps.Store), - users: storage.Typed[userEntry](deps.Store), - } + c := newCounter(deps.Store) return modules.Module{ CommandHook: c.Inc, Commands: []modules.Command{ diff --git a/internal/modules/stats/stats_test.go b/internal/modules/stats/stats_test.go index 8dd60c1..dc8afe2 100644 --- a/internal/modules/stats/stats_test.go +++ b/internal/modules/stats/stats_test.go @@ -11,16 +11,14 @@ import ( "github.com/tiennm99/miti99bot/internal/modules" "github.com/tiennm99/miti99bot/internal/storage" + "github.com/tiennm99/miti99bot/internal/systemstate" "github.com/tiennm99/miti99bot/internal/testutil" ) // newStatsCounter returns a fresh counter backed by an in-memory collection. func newStatsCounter() *counter { col := storage.NewMemoryProvider().Collection("stats") - return &counter{ - counts: storage.Typed[countEntry](col), - users: storage.Typed[userEntry](col), - } + return newCounter(col) } func TestNew_RegistersExpectedCommands(t *testing.T) { @@ -57,99 +55,97 @@ func updateFrom(id int64, username string) *models.Update { func TestInc_PersistsCountInStore(t *testing.T) { ctx := context.Background() - c := newStatsCounter() + provider := storage.NewMemoryProvider() + c := newCounter(provider.Collection("stats")) + docs := storage.Typed[usageEntry](provider.Collection("stats")) c.Inc(ctx, "ping", nil) c.Inc(ctx, "ping", nil) c.Inc(ctx, "wordle", nil) - entry, _, err := c.counts.Get(ctx, countKey("ping")) + entry, _, err := docs.Get(ctx, usageKey("ping", 0)) if err != nil { t.Fatalf("Get ping: %v", err) } - if entry.N != 2 { - t.Errorf("ping count = %d, want 2", entry.N) + if entry.Cmd != "ping" || entry.N != 2 || entry.UserID != 0 || entry.Username != "" { + t.Errorf("ping entry = %+v, want anonymous ping count 2", entry) } - entry2, _, err := c.counts.Get(ctx, countKey("wordle")) + entry2, _, err := docs.Get(ctx, usageKey("wordle", 0)) if err != nil { t.Fatalf("Get wordle: %v", err) } - if entry2.N != 1 { - t.Errorf("wordle count = %d, want 1", entry2.N) + if entry2.Cmd != "wordle" || entry2.N != 1 { + t.Errorf("wordle entry = %+v, want count 1", entry2) } } -func TestInc_WithUsernameWritesAllThreeKeys(t *testing.T) { +func TestInc_WithUsernameWritesPairEntry(t *testing.T) { ctx := context.Background() - c := newStatsCounter() + provider := storage.NewMemoryProvider() + c := newCounter(provider.Collection("stats")) + docs := storage.Typed[usageEntry](provider.Collection("stats")) c.Inc(ctx, "ping", updateFrom(42, "alice")) c.Inc(ctx, "ping", updateFrom(42, "alice")) - ce, _, err := c.counts.Get(ctx, countKey("ping")) + entry, _, err := docs.Get(ctx, usageKey("ping", 42)) if err != nil { - t.Fatalf("count:ping: %v", err) + t.Fatalf("ping:42: %v", err) } - if ce.N != 2 { - t.Errorf("count:ping N = %d, want 2", ce.N) + if entry.Cmd != "ping" || entry.UserID != 42 || entry.Username != "alice" || entry.N != 2 { + t.Errorf("ping:42 = %+v, want alice pair count 2", entry) } - ue, _, err := c.users.Get(ctx, userKey(42)) - if err != nil { - t.Fatalf("user:42: %v", err) - } - if ue.N != 2 || ue.Username != "alice" { - t.Errorf("user:42 = {%q, %d}, want {alice, 2}", ue.Username, ue.N) - } - - pe, _, err := c.counts.Get(ctx, pairKey("ping", 42)) - if err != nil { - t.Fatalf("pair:ping:42: %v", err) - } - if pe.N != 2 { - t.Errorf("pair:ping:42 N = %d, want 2", pe.N) + if _, _, err := docs.Get(ctx, usageKey("ping", 0)); !errors.Is(err, storage.ErrNotFound) { + t.Errorf("named user should not create anonymous ping bucket, got err=%v", err) } } func TestInc_EmptyUsernameSkipsUserAndPair(t *testing.T) { ctx := context.Background() - c := newStatsCounter() + provider := storage.NewMemoryProvider() + c := newCounter(provider.Collection("stats")) + docs := storage.Typed[usageEntry](provider.Collection("stats")) c.Inc(ctx, "ping", updateFrom(42, "")) - ce, _, err := c.counts.Get(ctx, countKey("ping")) + entry, _, err := docs.Get(ctx, usageKey("ping", 0)) if err != nil { - t.Fatalf("count:ping: %v", err) + t.Fatalf("ping: %v", err) } - if ce.N != 1 { - t.Errorf("count:ping N = %d, want 1", ce.N) + if entry.N != 1 || entry.UserID != 0 || entry.Username != "" { + t.Errorf("anonymous ping entry = %+v, want count 1 without user", entry) } - if _, _, err := c.users.Get(ctx, userKey(42)); !errors.Is(err, storage.ErrNotFound) { - t.Errorf("user:42 should be absent, got err=%v", err) - } - if _, _, err := c.counts.Get(ctx, pairKey("ping", 42)); !errors.Is(err, storage.ErrNotFound) { - t.Errorf("pair:ping:42 should be absent, got err=%v", err) + if _, _, err := docs.Get(ctx, usageKey("ping", 42)); !errors.Is(err, storage.ErrNotFound) { + t.Errorf("ping:42 should be absent, got err=%v", err) } } func TestInc_RefreshesUsernameOnRename(t *testing.T) { ctx := context.Background() - c := newStatsCounter() + provider := storage.NewMemoryProvider() + c := newCounter(provider.Collection("stats")) + docs := storage.Typed[usageEntry](provider.Collection("stats")) c.Inc(ctx, "ping", updateFrom(42, "alice")) + c.Inc(ctx, "wordle", updateFrom(42, "alice")) c.Inc(ctx, "ping", updateFrom(42, "alice2")) - ue, _, err := c.users.Get(ctx, userKey(42)) + ping, _, err := docs.Get(ctx, usageKey("ping", 42)) if err != nil { - t.Fatalf("user:42: %v", err) + t.Fatalf("ping:42: %v", err) } - if ue.Username != "alice2" { - t.Errorf("user:42 Username = %q, want %q", ue.Username, "alice2") + if ping.Username != "alice2" || ping.N != 2 { + t.Errorf("ping:42 = %+v, want alice2 count 2", ping) } - if ue.N != 2 { - t.Errorf("user:42 N = %d, want 2", ue.N) + wordle, _, err := docs.Get(ctx, usageKey("wordle", 42)) + if err != nil { + t.Fatalf("wordle:42: %v", err) + } + if wordle.Username != "alice2" || wordle.N != 1 { + t.Errorf("wordle:42 = %+v, want refreshed alice2 count 1", wordle) } } @@ -232,13 +228,98 @@ func TestCommandHook_FiredThroughModulesBuild(t *testing.T) { reg.RunCommandHooks(ctx, "ping", nil) // Verify by reading back via typed store (the same collection Build gave to stats). - statsStore := storage.Typed[countEntry](provider.Collection("stats")) - entry, _, err := statsStore.Get(ctx, countKey("ping")) + statsStore := storage.Typed[usageEntry](provider.Collection("stats")) + entry, _, err := statsStore.Get(ctx, usageKey("ping", 0)) if err != nil { - t.Fatalf("expected count:ping in store after hook: %v", err) + t.Fatalf("expected ping in store after hook: %v", err) } if entry.N != 1 { - t.Errorf("count:ping = %d, want 1", entry.N) + t.Errorf("ping = %d, want 1", entry.N) + } +} + +func TestInitStore_MigratesLegacyStatsOnceAndDeletesOldKeys(t *testing.T) { + ctx := context.Background() + provider := storage.NewMemoryProvider() + statsColl := provider.Collection("stats") + systemColl := provider.Collection(systemstate.CollectionName) + legacyCounts := storage.Typed[legacyCountEntry](statsColl) + legacyUsers := storage.Typed[legacyUserEntry](statsColl) + + if err := legacyCounts.Put(ctx, legacyCountPrefix+"ping", legacyCountEntry{N: 6}); err != nil { + t.Fatalf("legacy count ping: %v", err) + } + if err := legacyCounts.Put(ctx, legacyCountPrefix+"wordle", legacyCountEntry{N: 2}); err != nil { + t.Fatalf("legacy count wordle: %v", err) + } + if err := legacyUsers.Put(ctx, legacyUserPrefix+"1", legacyUserEntry{Username: "alice", N: 4}); err != nil { + t.Fatalf("legacy user alice: %v", err) + } + if err := legacyUsers.Put(ctx, legacyUserPrefix+"2", legacyUserEntry{Username: "bob", N: 3}); err != nil { + t.Fatalf("legacy user bob: %v", err) + } + if err := legacyCounts.Put(ctx, legacyPairPrefix+"ping:1", legacyCountEntry{N: 3}); err != nil { + t.Fatalf("legacy pair ping alice: %v", err) + } + if err := legacyCounts.Put(ctx, legacyPairPrefix+"ping:2", legacyCountEntry{N: 2}); err != nil { + t.Fatalf("legacy pair ping bob: %v", err) + } + if err := legacyCounts.Put(ctx, legacyPairPrefix+"wordle:2", legacyCountEntry{N: 2}); err != nil { + t.Fatalf("legacy pair wordle bob: %v", err) + } + + if err := InitStore(ctx, statsColl, systemColl); err != nil { + t.Fatalf("InitStore: %v", err) + } + + usageDocs := storage.Typed[usageEntry](statsColl) + assertUsageEntry(t, usageDocs, usageKey("ping", 1), usageEntry{Cmd: "ping", UserID: 1, Username: "alice", N: 3}) + assertUsageEntry(t, usageDocs, usageKey("ping", 2), usageEntry{Cmd: "ping", UserID: 2, Username: "bob", N: 2}) + assertUsageEntry(t, usageDocs, usageKey("wordle", 2), usageEntry{Cmd: "wordle", UserID: 2, Username: "bob", N: 2}) + assertUsageEntry(t, usageDocs, usageKey("ping", 0), usageEntry{Cmd: "ping", N: 1}) + + for _, key := range []string{ + legacyCountPrefix + "ping", + legacyCountPrefix + "wordle", + legacyPairPrefix + "ping:1", + legacyPairPrefix + "ping:2", + legacyPairPrefix + "wordle:2", + legacyUserPrefix + "1", + legacyUserPrefix + "2", + } { + if _, _, err := legacyCounts.Get(ctx, key); !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("legacy key %s should be deleted, got err=%v", key, err) + } + } + + sys := systemstate.New(systemColl) + rec, ok, err := sys.Get(ctx, usageMigrationKey) + if err != nil || !ok { + t.Fatalf("migration marker ok=%v err=%v", ok, err) + } + if rec.Status != "done" || rec.Count != 4 { + t.Fatalf("migration marker = %+v, want done count 4", rec) + } + + if err := legacyCounts.Put(ctx, legacyCountPrefix+"coin", legacyCountEntry{N: 99}); err != nil { + t.Fatalf("legacy count after marker: %v", err) + } + if err := InitStore(ctx, statsColl, systemColl); err != nil { + t.Fatalf("InitStore second run: %v", err) + } + if _, _, err := usageDocs.Get(ctx, usageKey("coin", 0)); !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("second InitStore should skip migration after marker, got err=%v", err) + } +} + +func assertUsageEntry(t *testing.T, docs storage.DocStore[usageEntry], key string, want usageEntry) { + t.Helper() + got, _, err := docs.Get(context.Background(), key) + if err != nil { + t.Fatalf("usage entry %s: %v", key, err) + } + if got != want { + t.Fatalf("usage entry %s = %+v, want %+v", key, got, want) } } diff --git a/internal/modules/stats/usage_store.go b/internal/modules/stats/usage_store.go new file mode 100644 index 0000000..dbf7bf5 --- /dev/null +++ b/internal/modules/stats/usage_store.go @@ -0,0 +1,451 @@ +package stats + +import ( + "context" + "errors" + "fmt" + "strconv" + "sync" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + + "github.com/tiennm99/miti99bot/internal/storage" +) + +func bsonField(key string, value any) bson.E { + return bson.E{Key: key, Value: value} +} + +// usageEntry is the only stats payload shape. One document represents either +// a command-level anonymous bucket or a concrete (command, Telegram user) pair. +type usageEntry struct { + Cmd string `json:"cmd" bson:"cmd"` + UserID int64 `json:"uid,omitempty" bson:"uid,omitempty"` + Username string `json:"user,omitempty" bson:"user,omitempty"` + N int64 `json:"n" bson:"n"` +} + +type usageUser struct { + ID int64 + Username string +} + +type usageStore interface { + Increment(ctx context.Context, cmd string, user usageUser, hasUser bool) error + TopCommands(ctx context.Context, limit int) ([]row, error) + TopUsers(ctx context.Context, limit int) ([]row, error) + CommandsByUser(ctx context.Context, username string, limit int) ([]row, bool, error) + UsersByCommand(ctx context.Context, cmd string, limit int) ([]row, error) +} + +func newUsageStore(coll storage.Collection) usageStore { + if mongoColl, ok := storage.MongoCollection(coll); ok { + return &mongoUsageStore{coll: mongoColl} + } + return &docUsageStore{docs: storage.Typed[usageEntry](coll)} +} + +func usageKey(cmd string, userID int64) string { + if userID == 0 { + return cmd + } + return cmd + ":" + strconv.FormatInt(userID, 10) +} + +type docUsageStore struct { + mu sync.Mutex + docs storage.DocStore[usageEntry] +} + +func (s *docUsageStore) Increment(ctx context.Context, cmd string, user usageUser, hasUser bool) error { + s.mu.Lock() + defer s.mu.Unlock() + + userID := int64(0) + if hasUser { + userID = user.ID + } + key := usageKey(cmd, userID) + entry, _, err := s.docs.Get(ctx, key) + switch { + case errors.Is(err, storage.ErrNotFound): + entry = usageEntry{} + case err != nil: + return fmt.Errorf("stats get %s: %w", key, err) + } + + entry.Cmd = cmd + entry.N++ + if hasUser { + entry.UserID = user.ID + entry.Username = user.Username + } else { + entry.UserID = 0 + entry.Username = "" + } + if err := s.docs.Put(ctx, key, entry); err != nil { + return fmt.Errorf("stats put %s: %w", key, err) + } + if hasUser { + return s.refreshUsernameLocked(ctx, user) + } + return nil +} + +func (s *docUsageStore) refreshUsernameLocked(ctx context.Context, user usageUser) error { + entries, err := s.loadEntriesLocked(ctx) + if err != nil { + return err + } + for _, e := range entries { + if e.UserID != user.ID || e.Username == user.Username { + continue + } + e.Username = user.Username + if err := s.docs.Put(ctx, usageKey(e.Cmd, e.UserID), e); err != nil { + return fmt.Errorf("stats refresh username %d: %w", user.ID, err) + } + } + return nil +} + +func (s *docUsageStore) TopCommands(ctx context.Context, limit int) ([]row, error) { + s.mu.Lock() + defer s.mu.Unlock() + + entries, err := s.loadEntriesLocked(ctx) + if err != nil { + return nil, err + } + totals := make(map[string]int64) + for _, e := range entries { + totals[e.Cmd] += e.N + } + rows := make([]row, 0, len(totals)) + for cmd, n := range totals { + rows = append(rows, row{display: "/" + cmd, n: n}) + } + sortRows(rows) + return limitRows(rows, limit), nil +} + +func (s *docUsageStore) TopUsers(ctx context.Context, limit int) ([]row, error) { + s.mu.Lock() + defer s.mu.Unlock() + + entries, err := s.loadEntriesLocked(ctx) + if err != nil { + return nil, err + } + type total struct { + username string + n int64 + } + totals := make(map[int64]total) + for _, e := range entries { + if e.UserID == 0 || e.Username == "" { + continue + } + t := totals[e.UserID] + t.username = e.Username + t.n += e.N + totals[e.UserID] = t + } + rows := make([]row, 0, len(totals)) + for _, t := range totals { + rows = append(rows, row{display: "@" + t.username, n: t.n}) + } + sortRows(rows) + return limitRows(rows, limit), nil +} + +func (s *docUsageStore) CommandsByUser(ctx context.Context, username string, limit int) ([]row, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + + entries, err := s.loadEntriesLocked(ctx) + if err != nil { + return nil, false, err + } + var ( + userID int64 + found bool + ) + for _, e := range entries { + if e.UserID != 0 && e.Username == username { + userID = e.UserID + found = true + break + } + } + if !found { + return nil, false, nil + } + + rows := make([]row, 0) + for _, e := range entries { + if e.UserID == userID { + rows = append(rows, row{display: "/" + e.Cmd, n: e.N}) + } + } + sortRows(rows) + return limitRows(rows, limit), true, nil +} + +func (s *docUsageStore) UsersByCommand(ctx context.Context, cmd string, limit int) ([]row, error) { + s.mu.Lock() + defer s.mu.Unlock() + + entries, err := s.loadEntriesLocked(ctx) + if err != nil { + return nil, err + } + rows := make([]row, 0) + for _, e := range entries { + if e.Cmd == cmd && e.UserID != 0 && e.Username != "" { + rows = append(rows, row{display: "@" + e.Username, n: e.N}) + } + } + sortRows(rows) + return limitRows(rows, limit), nil +} + +func (s *docUsageStore) loadEntriesLocked(ctx context.Context) ([]usageEntry, error) { + keys, err := s.docs.List(ctx, "") + if err != nil { + return nil, fmt.Errorf("stats list: %w", err) + } + entries := make([]usageEntry, 0, len(keys)) + for _, key := range keys { + entry, _, err := s.docs.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("stats get %s: %w", key, err) + } + if entry.Cmd == "" { + continue + } + entries = append(entries, entry) + } + return entries, nil +} + +type mongoUsageStore struct { + coll *mongo.Collection +} + +func (s *mongoUsageStore) Increment(ctx context.Context, cmd string, user usageUser, hasUser bool) error { + userID := int64(0) + if hasUser { + userID = user.ID + } + + set := bson.M{"cmd": cmd} + update := bson.M{ + "$set": set, + "$inc": bson.M{"n": int64(1), "version": int64(1)}, + "$currentDate": bson.M{"updatedAt": true}, + } + if hasUser { + set["uid"] = user.ID + set["user"] = user.Username + } else { + update["$unset"] = bson.M{"uid": "", "user": ""} + } + + key := usageKey(cmd, userID) + if _, err := s.coll.UpdateOne(ctx, + bson.M{"_id": key}, + update, + options.UpdateOne().SetUpsert(true), + ); err != nil { + return fmt.Errorf("mongo stats increment %s: %w", key, err) + } + if !hasUser { + return nil + } + if _, err := s.coll.UpdateMany(ctx, + bson.M{"uid": user.ID, "user": bson.M{"$ne": user.Username}}, + bson.M{ + "$set": bson.M{"user": user.Username}, + "$inc": bson.M{"version": int64(1)}, + "$currentDate": bson.M{"updatedAt": true}, + }, + ); err != nil { + return fmt.Errorf("mongo stats refresh username %d: %w", user.ID, err) + } + return nil +} + +func (s *mongoUsageStore) TopCommands(ctx context.Context, limit int) ([]row, error) { + pipeline := mongo.Pipeline{ + bson.D{bsonField("$match", bson.D{bsonField("cmd", bson.D{bsonField("$type", "string")})})}, + bson.D{bsonField("$group", bson.D{bsonField("_id", "$cmd"), bsonField("n", bson.D{bsonField("$sum", "$n")})})}, + bson.D{bsonField("$sort", bson.D{bsonField("n", -1), bsonField("_id", 1)})}, + } + pipeline = withLimit(pipeline, limit) + + cur, err := s.coll.Aggregate(ctx, pipeline) + if err != nil { + return nil, fmt.Errorf("mongo stats top commands: %w", err) + } + defer func() { _ = cur.Close(ctx) }() + + var rows []row + for cur.Next(ctx) { + var doc struct { + Cmd string `bson:"_id"` + N int64 `bson:"n"` + } + if err := cur.Decode(&doc); err != nil { + return nil, fmt.Errorf("mongo stats top commands decode: %w", err) + } + rows = append(rows, row{display: "/" + doc.Cmd, n: doc.N}) + } + if err := cur.Err(); err != nil { + return nil, fmt.Errorf("mongo stats top commands cursor: %w", err) + } + return rows, nil +} + +func (s *mongoUsageStore) TopUsers(ctx context.Context, limit int) ([]row, error) { + pipeline := mongo.Pipeline{ + bson.D{bsonField("$match", bson.D{ + bsonField("uid", bson.D{bsonField("$gt", int64(0))}), + bsonField("user", bson.D{bsonField("$type", "string"), bsonField("$ne", "")}), + })}, + bson.D{bsonField("$sort", bson.D{bsonField("updatedAt", -1)})}, + bson.D{bsonField("$group", bson.D{ + bsonField("_id", "$uid"), + bsonField("user", bson.D{bsonField("$first", "$user")}), + bsonField("n", bson.D{bsonField("$sum", "$n")}), + })}, + bson.D{bsonField("$sort", bson.D{bsonField("n", -1), bsonField("user", 1)})}, + } + pipeline = withLimit(pipeline, limit) + + cur, err := s.coll.Aggregate(ctx, pipeline) + if err != nil { + return nil, fmt.Errorf("mongo stats top users: %w", err) + } + defer func() { _ = cur.Close(ctx) }() + + var rows []row + for cur.Next(ctx) { + var doc struct { + User string `bson:"user"` + N int64 `bson:"n"` + } + if err := cur.Decode(&doc); err != nil { + return nil, fmt.Errorf("mongo stats top users decode: %w", err) + } + rows = append(rows, row{display: "@" + doc.User, n: doc.N}) + } + if err := cur.Err(); err != nil { + return nil, fmt.Errorf("mongo stats top users cursor: %w", err) + } + return rows, nil +} + +func (s *mongoUsageStore) CommandsByUser(ctx context.Context, username string, limit int) ([]row, bool, error) { + userID, found, err := s.userIDByUsername(ctx, username) + if err != nil || !found { + return nil, found, err + } + + opts := options.Find(). + SetProjection(bson.M{"cmd": 1, "n": 1}). + SetSort(bson.D{bsonField("n", -1), bsonField("cmd", 1)}) + if limit > 0 { + opts.SetLimit(int64(limit)) + } + cur, err := s.coll.Find(ctx, bson.M{ + "uid": userID, + "cmd": bson.M{"$type": "string"}, + }, opts) + if err != nil { + return nil, true, fmt.Errorf("mongo stats commands by user %s: %w", username, err) + } + defer func() { _ = cur.Close(ctx) }() + + var rows []row + for cur.Next(ctx) { + var doc usageEntry + if err := cur.Decode(&doc); err != nil { + return nil, true, fmt.Errorf("mongo stats commands by user decode: %w", err) + } + rows = append(rows, row{display: "/" + doc.Cmd, n: doc.N}) + } + if err := cur.Err(); err != nil { + return nil, true, fmt.Errorf("mongo stats commands by user cursor: %w", err) + } + return rows, true, nil +} + +func (s *mongoUsageStore) UsersByCommand(ctx context.Context, cmd string, limit int) ([]row, error) { + opts := options.Find(). + SetProjection(bson.M{"user": 1, "n": 1}). + SetSort(bson.D{bsonField("n", -1), bsonField("user", 1)}) + if limit > 0 { + opts.SetLimit(int64(limit)) + } + cur, err := s.coll.Find(ctx, bson.M{ + "cmd": cmd, + "uid": bson.M{"$gt": int64(0)}, + "user": bson.M{"$type": "string", "$ne": ""}, + }, opts) + if err != nil { + return nil, fmt.Errorf("mongo stats users by command %s: %w", cmd, err) + } + defer func() { _ = cur.Close(ctx) }() + + var rows []row + for cur.Next(ctx) { + var doc usageEntry + if err := cur.Decode(&doc); err != nil { + return nil, fmt.Errorf("mongo stats users by command decode: %w", err) + } + rows = append(rows, row{display: "@" + doc.Username, n: doc.N}) + } + if err := cur.Err(); err != nil { + return nil, fmt.Errorf("mongo stats users by command cursor: %w", err) + } + return rows, nil +} + +func (s *mongoUsageStore) userIDByUsername(ctx context.Context, username string) (int64, bool, error) { + var doc struct { + UserID int64 `bson:"uid"` + } + err := s.coll.FindOne(ctx, + bson.M{ + "uid": bson.M{"$gt": int64(0)}, + "user": username, + }, + options.FindOne(). + SetProjection(bson.M{"uid": 1}). + SetSort(bson.D{bsonField("updatedAt", -1)}), + ).Decode(&doc) + if err != nil { + if errors.Is(err, mongo.ErrNoDocuments) { + return 0, false, nil + } + return 0, false, fmt.Errorf("mongo stats find user %s: %w", username, err) + } + return doc.UserID, true, nil +} + +func withLimit(pipeline mongo.Pipeline, limit int) mongo.Pipeline { + if limit <= 0 { + return pipeline + } + return append(pipeline, bson.D{bsonField("$limit", int64(limit))}) +} + +func limitRows(rows []row, limit int) []row { + if limit <= 0 || len(rows) <= limit { + return rows + } + return rows[:limit] +} diff --git a/internal/modules/stats/views.go b/internal/modules/stats/views.go index 060d50c..6ccf8a6 100644 --- a/internal/modules/stats/views.go +++ b/internal/modules/stats/views.go @@ -4,9 +4,7 @@ import ( "context" "fmt" "sort" - "strconv" "strings" - "sync" "github.com/go-telegram/bot" "github.com/go-telegram/bot/models" @@ -89,225 +87,56 @@ func parseSubargs(update *models.Update) string { } func viewTopCommands(ctx context.Context, c *counter) string { - keys, err := c.counts.List(ctx, countPrefix) + rows, err := c.store.TopCommands(ctx, topK) if err != nil { - log.Error("stats: store list failed", "err", err) + log.Error("stats: top commands failed", "err", err) return "Could not load stats. Try again later." } - if len(keys) == 0 { - return "No command stats yet." - } - rows := fanOutCountRows(ctx, c, keys, func(k string) string { - return "/" + strings.TrimPrefix(k, countPrefix) - }) if len(rows) == 0 { return "No command stats yet." } - sortRows(rows) return renderTopN("Command usage:", rows, topK) } func viewTopUsers(ctx context.Context, c *counter) string { - users := loadUserRowsWithID(ctx, c) - rows := make([]row, 0, len(users)) - for _, u := range users { - if u.Username == "" { - continue - } - rows = append(rows, row{display: "@" + u.Username, n: u.N}) + rows, err := c.store.TopUsers(ctx, topK) + if err != nil { + log.Error("stats: top users failed", "err", err) + return "Could not load stats. Try again later." } if len(rows) == 0 { return "No user stats yet." } - sortRows(rows) return renderTopN("Top users:", rows, topK) } func viewUserCommands(ctx context.Context, c *counter, username string) string { - users := loadUserRowsWithID(ctx, c) - // Telegram allows username reuse after one user changes theirs, so two - // distinct user IDs may briefly share a username. First match wins; map - // iteration order is nondeterministic, so the choice is not stable across - // renders. Accepted limitation — disambiguating would need historical - // (timestamp, ID) data we don't store. - var ( - foundID int64 - ok bool - ) - for id, u := range users { - if u.Username == username { - foundID = id - ok = true - break - } - } - if !ok { - return fmt.Sprintf("User @%s not found.", username) - } - - keys, err := c.counts.List(ctx, pairPrefix) + rows, found, err := c.store.CommandsByUser(ctx, username, topK) if err != nil { - log.Error("stats: store list failed", "err", err) + log.Error("stats: commands by user failed", "username", username, "err", err) return "Could not load stats. Try again later." } - // Leading colon disambiguates: ":2" does not match a key ending in "12" - // because IDs are bare decimal integers with no internal punctuation. - suffix := ":" + strconv.FormatInt(foundID, 10) - var matching []string - for _, k := range keys { - if strings.HasSuffix(k, suffix) { - matching = append(matching, k) - } + if !found { + return fmt.Sprintf("User @%s not found.", username) } - if len(matching) == 0 { - return fmt.Sprintf("No commands recorded for @%s.", username) - } - - rows := fanOutCountRows(ctx, c, matching, func(k string) string { - // k looks like "pair::" — drop the prefix and trailing :. - rest := strings.TrimPrefix(k, pairPrefix) - idx := strings.LastIndexByte(rest, ':') - if idx < 0 { - return "/" + rest - } - return "/" + rest[:idx] - }) if len(rows) == 0 { return fmt.Sprintf("No commands recorded for @%s.", username) } - sortRows(rows) return renderTopN(fmt.Sprintf("Commands by @%s:", username), rows, topK) } func viewCmdUsers(ctx context.Context, c *counter, cmd string) string { - prefix := pairPrefix + cmd + ":" - keys, err := c.counts.List(ctx, prefix) + rows, err := c.store.UsersByCommand(ctx, cmd, topK) if err != nil { - log.Error("stats: store list failed", "err", err) + log.Error("stats: users by command failed", "command", cmd, "err", err) return "Could not load stats. Try again later." } - if len(keys) == 0 { + if len(rows) == 0 { return fmt.Sprintf("Command /%s has no users yet.", cmd) } - - type result struct { - r row - ok bool - } - results := make([]result, len(keys)) - var wg sync.WaitGroup - for i, k := range keys { - wg.Add(1) - go func(i int, k string) { - defer wg.Done() - ce, _, err := c.counts.Get(ctx, k) - if err != nil { - log.Error("stats: store get failed", "key", k, "err", err) - return - } - idStr := strings.TrimPrefix(k, prefix) - id, err := strconv.ParseInt(idStr, 10, 64) - if err != nil { - return - } - ue, _, err := c.users.Get(ctx, userKey(id)) - if err != nil || ue.Username == "" { - return - } - results[i] = result{r: row{display: "@" + ue.Username, n: ce.N}, ok: true} - }(i, k) - } - wg.Wait() - - rows := make([]row, 0, len(results)) - for _, r := range results { - if r.ok { - rows = append(rows, r.r) - } - } - if len(rows) == 0 { - return fmt.Sprintf("Command /%s has no named users yet.", cmd) - } - sortRows(rows) return renderTopN(fmt.Sprintf("Users of /%s:", cmd), rows, topK) } -// fanOutCountRows runs Get on each key in parallel. displayFor maps the -// raw sort key to the human-readable label that gets rendered. Missing / -// errored keys are skipped (logged at error level). -func fanOutCountRows(ctx context.Context, c *counter, keys []string, displayFor func(k string) string) []row { - type result struct { - r row - ok bool - } - results := make([]result, len(keys)) - var wg sync.WaitGroup - for i, k := range keys { - wg.Add(1) - go func(i int, k string) { - defer wg.Done() - entry, _, err := c.counts.Get(ctx, k) - if err != nil { - log.Error("stats: store get failed during render", "key", k, "err", err) - return - } - results[i] = result{r: row{display: displayFor(k), n: entry.N}, ok: true} - }(i, k) - } - wg.Wait() - - rows := make([]row, 0, len(results)) - for _, r := range results { - if r.ok { - rows = append(rows, r.r) - } - } - return rows -} - -// loadUserRowsWithID returns userID → userEntry for every user:* row. Used by -// viewTopUsers (renders) and viewUserCommands (resolves a username to its ID). -func loadUserRowsWithID(ctx context.Context, c *counter) map[int64]userEntry { - keys, err := c.users.List(ctx, userPrefix) - if err != nil { - log.Error("stats: store list failed", "err", err) - return nil - } - type pair struct { - id int64 - entry userEntry - ok bool - } - results := make([]pair, len(keys)) - var wg sync.WaitGroup - for i, k := range keys { - wg.Add(1) - go func(i int, k string) { - defer wg.Done() - idStr := strings.TrimPrefix(k, userPrefix) - id, err := strconv.ParseInt(idStr, 10, 64) - if err != nil { - return - } - entry, _, err := c.users.Get(ctx, k) - if err != nil { - log.Error("stats: store get failed", "key", k, "err", err) - return - } - results[i] = pair{id: id, entry: entry, ok: true} - }(i, k) - } - wg.Wait() - - out := make(map[int64]userEntry, len(results)) - for _, p := range results { - if p.ok { - out[p.id] = p.entry - } - } - return out -} - func sortRows(rows []row) { sort.Slice(rows, func(i, j int) bool { if rows[i].n != rows[j].n { diff --git a/internal/storage/mongodb_provider.go b/internal/storage/mongodb_provider.go index 06da8fa..a195085 100644 --- a/internal/storage/mongodb_provider.go +++ b/internal/storage/mongodb_provider.go @@ -27,3 +27,14 @@ func (p *MongoProvider) Collection(moduleName string) Collection { } return mongoCollection{coll: p.db.Collection(moduleName), module: moduleName} } + +// MongoCollection returns the native MongoDB collection behind c when the +// active backend is MongoDB. Modules with query-shaped data can opt into Mongo +// operators while still keeping the memory backend for tests/local runs. +func MongoCollection(c Collection) (*mongo.Collection, bool) { + h, ok := c.(mongoCollection) + if !ok { + return nil, false + } + return h.coll, true +} diff --git a/internal/systemstate/systemstate.go b/internal/systemstate/systemstate.go new file mode 100644 index 0000000..6ecc482 --- /dev/null +++ b/internal/systemstate/systemstate.go @@ -0,0 +1,46 @@ +package systemstate + +import ( + "context" + "errors" + + "github.com/tiennm99/miti99bot/internal/storage" +) + +// CollectionName is the app-level collection for startup tasks and other +// process metadata that does not belong to a feature module. +const CollectionName = "system" + +// Record is intentionally small and generic. Stable keys carry the meaning; +// optional fields let startup tasks record counts and completion state. +type Record struct { + Kind string `json:"kind" bson:"kind"` + Name string `json:"name" bson:"name"` + Status string `json:"status,omitempty" bson:"status,omitempty"` + Count int64 `json:"count,omitempty" bson:"count,omitempty"` + CompletedAt int64 `json:"completed_at,omitempty" bson:"completed_at,omitempty"` + UpdatedAt int64 `json:"updated_at" bson:"updated_at"` +} + +type Store struct { + docs storage.DocStore[Record] +} + +func New(coll storage.Collection) Store { + return Store{docs: storage.Typed[Record](coll)} +} + +func (s Store) Get(ctx context.Context, key string) (Record, bool, error) { + rec, _, err := s.docs.Get(ctx, key) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + return Record{}, false, nil + } + return Record{}, false, err + } + return rec, true, nil +} + +func (s Store) Put(ctx context.Context, key string, rec Record) error { + return s.docs.Put(ctx, key, rec) +} diff --git a/plans/reports/260701-1000-mongodb-stats-design-research.md b/plans/reports/260701-1000-mongodb-stats-design-research.md new file mode 100644 index 0000000..e98030f --- /dev/null +++ b/plans/reports/260701-1000-mongodb-stats-design-research.md @@ -0,0 +1,265 @@ +--- +type: research +topic: mongodb-stats-design +created_at: 2026-07-01T03:00:06Z +--- + +# Research Report: MongoDB Stats Design + +## Executive Summary + +For this bot's `/stats` feature, best primary model is one aggregate document per +`(command, user)` pair, plus one anonymous command bucket when sender has no +username. This keeps documents bounded, uses MongoDB atomic `$inc`, and makes all +current views queryable without dynamic object keys. + +Do not store stats as "one user document with commands map" or "one command +document with users map" as primary shape. Those designs make one query easy and +the opposite query expensive, create hot growing documents, and push us toward +dynamic field names. Pair docs are boring and better. + +Recommended document: + +```js +{ _id: "ping:42", cmd: "ping", uid: 42, user: "alice", n: 3 } +{ _id: "ping", cmd: "ping", n: 2 } // no public username +``` + +## Methodology + +- Sources consulted: MongoDB official docs, Go driver docs, local codebase. +- Date: 2026-07-01. +- Key terms: MongoDB schema design, unbounded arrays, `$inc`, atomicity, + aggregation `$group`, `$sort`, compound indexes, ESR guideline. +- Scope: stats module only. Bot persistence remains one collection per module. + +## Codebase Context + +- Repo: Go Telegram bot. +- Storage: `internal/storage` has one MongoDB collection per module. +- Current stats: `count:*`, `user:*`, `pair:*` key families in one collection. +- Current views: + - `/stats`: top commands. + - `/stats users`: top users. + - `/stats user `: commands by user. + - `/stats cmd `: users by command. +- Mongo docs already store flattened BSON fields, so stats fields can be queried. + +## Key Findings + +### 1. Avoid Growing Nested User/Command Maps + +MongoDB docs warn against unbounded arrays because growing documents hurt +performance and risk document size limits. The same practical concern applies to +ever-growing embedded maps like `commands` under a user or `users` under a +command: one document grows forever and becomes a write hotspot. + +Source: MongoDB "Avoid Unbounded Arrays": +https://www.mongodb.com/docs/manual/data-modeling/design-antipatterns/unbounded-arrays/ + +### 2. Use Atomic `$inc` For Counts + +MongoDB `$inc` creates missing numeric fields, increments by the given amount, +and is atomic within one document. This fits per-pair counter documents exactly. + +Source: MongoDB `$inc` operator: +https://www.mongodb.com/docs/manual/reference/operator/update/inc/ + +MongoDB write operations are atomic at single-document level. Multi-document +updates are not atomic as a whole, so each counter increment should update one +aggregate document. + +Source: MongoDB atomicity: +https://www.mongodb.com/docs/manual/core/write-operations-atomicity/ + +### 3. Use Aggregation For Cross-Dimension Totals + +MongoDB `$group` combines documents by a group key. That fits: + +```js +// top commands +[{ $group: { _id: "$cmd", n: { $sum: "$n" } } }] + +// top users +[{ $match: { uid: { $gt: 0 } } }, + { $group: { _id: "$uid", user: { $first: "$user" }, n: { $sum: "$n" } } }] +``` + +Source: MongoDB `$group`: +https://www.mongodb.com/docs/manual/reference/operator/aggregation/group/ + +Use `$sort` then `$limit` for top K. MongoDB can coalesce adjacent sort/limit so +only top N must be kept in memory. + +Source: MongoDB `$sort` optimization: +https://www.mongodb.com/docs/manual/reference/operator/aggregation/sort/ + +### 4. Index From Query Shape, Not Guesswork + +MongoDB aggregation can use indexes, especially early `$match` and `$sort` +stages. The ESR guideline says equality fields first, then sort, then range for +compound indexes. + +Sources: +- Aggregation pipeline optimization: + https://www.mongodb.com/docs/manual/core/aggregation-pipeline-optimization/ +- ESR guideline: + https://www.mongodb.com/docs/manual/tutorial/equality-sort-range-guideline/ +- Indexing strategies: + https://www.mongodb.com/docs/manual/applications/indexes/ + +For this module, start with no extra indexes or a tiny set. Traffic likely low. +Add indexes once data grows or `/stats` gets slow. + +Useful indexes if needed: + +```js +db.stats.createIndex({ cmd: 1, n: -1 }) // /stats cmd +db.stats.createIndex({ uid: 1, n: -1 }) // /stats user +db.stats.createIndex({ user: 1, updatedAt: -1 }) // resolve username +``` + +Do not add many indexes early. Each index costs memory/disk and write work. + +## Comparative Analysis + +| Shape | Example | Good | Bad | Verdict | +|---|---|---|---|---| +| User-rooted | `{ uid, user, commands: { ping: 3 } }` | `/stats user` easy | `/stats cmd` scans users, dynamic keys, growing doc | Reject | +| Command-rooted | `{ cmd, users: { "42": { n: 3 } } }` | `/stats cmd` easy | `/stats user` scans commands, hot command docs, dynamic keys | Reject | +| Pair-rooted | `{ cmd, uid, user, n }` | All 4 views queryable, bounded docs, atomic `$inc` | Aggregation needed for totals | Recommend | +| Event log | `{ cmd, uid, user, at }` per invocation | Full history | Too much data, needs rollups | Overkill | + +## Implementation Recommendations + +### Primary Model + +Use one document per aggregate: + +```go +type usageEntry struct { + Cmd string `bson:"cmd" json:"cmd"` + UserID int64 `bson:"uid,omitempty" json:"uid,omitempty"` + Username string `bson:"user,omitempty" json:"user,omitempty"` + N int64 `bson:"n" json:"n"` +} +``` + +Keys: + +```text + anonymous/no-public-username bucket +: named user pair bucket +``` + +### Write Path + +Use MongoDB `UpdateOne` with `upsert: true`: + +```js +db.stats.updateOne( + { _id: "ping:42" }, + { + $set: { cmd: "ping", uid: 42, user: "alice" }, + $inc: { n: 1 }, + $currentDate: { updatedAt: true } + }, + { upsert: true } +) +``` + +This avoids read-modify-write races in the current generic `DocStore` loop. + +### Read Path + +Use Mongo queries/aggregation when backend is MongoDB. Keep in-memory fallback +for tests and no-database local run. + +Queries: + +```js +// /stats +db.stats.aggregate([ + { $group: { _id: "$cmd", n: { $sum: "$n" } } }, + { $sort: { n: -1, _id: 1 } }, + { $limit: 20 } +]) + +// /stats users +db.stats.aggregate([ + { $match: { uid: { $gt: 0 }, user: { $type: "string", $ne: "" } } }, + { $group: { _id: "$uid", user: { $first: "$user" }, n: { $sum: "$n" } } }, + { $sort: { n: -1, user: 1 } }, + { $limit: 20 } +]) + +// /stats user alice +db.stats.find({ uid: 42 }).sort({ n: -1, cmd: 1 }).limit(20) + +// /stats cmd ping +db.stats.find({ cmd: "ping", uid: { $gt: 0 } }).sort({ n: -1, user: 1 }).limit(20) +``` + +### Username Rename + +When a user calls a command with a new username, refresh all docs for that `uid` +with `UpdateMany({ uid }, { $set: { user } })`. It is not atomic as a whole, but +stats are best-effort and eventual consistency is acceptable. + +### Migration + +Existing prod data likely in old key prefixes. Options: + +1. No migration: simplest, stats reset after deploy. +2. Read old + new for one release: more code, little value. +3. One-off migration script: only if historical stats matter. + +Recommendation: no migration unless user explicitly wants historical stats. + +## Common Pitfalls + +- Do not use dynamic fields like `commands.ping` or `users.42` as primary model. +- Do not grow one document forever. +- Do not use generic get-then-put for hot counters on MongoDB. +- Do not over-index small collections. +- Do not use username as identity. Telegram username can change; use `uid`. + +## Decision + +Recommended: pair-rooted aggregate docs, Mongo atomic upsert increments, Mongo +aggregation for display views, memory fallback for tests. + +## References + +- MongoDB Avoid Unbounded Arrays: + https://www.mongodb.com/docs/manual/data-modeling/design-antipatterns/unbounded-arrays/ +- MongoDB `$inc`: + https://www.mongodb.com/docs/manual/reference/operator/update/inc/ +- MongoDB Atomicity: + https://www.mongodb.com/docs/manual/core/write-operations-atomicity/ +- MongoDB `$group`: + https://www.mongodb.com/docs/manual/reference/operator/aggregation/group/ +- MongoDB `$sort`: + https://www.mongodb.com/docs/manual/reference/operator/aggregation/sort/ +- MongoDB Aggregation Optimization: + https://www.mongodb.com/docs/manual/core/aggregation-pipeline-optimization/ +- MongoDB ESR Guideline: + https://www.mongodb.com/docs/manual/tutorial/equality-sort-range-guideline/ +- MongoDB Indexing Strategies: + https://www.mongodb.com/docs/manual/applications/indexes/ +- MongoDB Go Driver Update: + https://www.mongodb.com/docs/drivers/go/current/crud/update/ +- MongoDB Go Driver Indexes: + https://www.mongodb.com/docs/drivers/go/current/indexes/ + +## Next Steps + +1. Confirm pair-rooted aggregate docs. +2. Finish stats implementation. +3. Run `go test ./internal/modules/stats ./internal/storage`. +4. Optional later: add Mongo integration test if local Mongo available. + +## Unresolved Questions + +- Keep old stats by migration, or accept reset? +- Add indexes now, or wait until data size/latency proves need?