tiennm99/miti99bot
1
0
Fork 0
mirror of https://github.com/tiennm99/miti99bot.git synced 2026-08-09 18:26:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(stats): use queryable Mongo usage records

Browse Source
This commit is contained in:
tiennm99
2026-07-01 10:22:03 +07:00
parent b05bbab3cd
commit 6fd65729ac
13 changed files with 1275 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+1 -1
View File
@@ -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 ------------------------------------------------------------
README.md
+3 -2
View File
@@ -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
cmd/server/main.go
+6
View File
@@ -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 {
docs/deploy-coolify-selfhosted.md
+8 -2
View File
@@ -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: <user key>, ...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
internal/modules/stats/startup.go
+215
View File
@@ -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()
}
internal/modules/stats/startup_mongo_test.go
+100
View File
@@ -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)
}
}
internal/modules/stats/stats.go
+23 -108
View File
@@ -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:<cmd> → countEntry — total per command
// user:<userID> → userEntry — cached username + total per user
// pair:<cmd>:<userID> → 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:<cmd>. When the originating user has a Telegram
// username, also increments user:<id> (refreshing the cached username) and
// pair:<cmd>:<id>. 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{
internal/modules/stats/stats_test.go
+133 -52
View File
@@ -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)
}
}
internal/modules/stats/usage_store.go
+451
View File
@@ -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]
}
internal/modules/stats/views.go
+13 -184
View File
@@ -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:<cmd>:<id>" — drop the prefix and trailing :<id>.
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 {
internal/storage/mongodb_provider.go
+11
View File
@@ -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
}
internal/systemstate/systemstate.go
+46
View File
@@ -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)
}
plans/reports/260701-1000-mongodb-stats-design-research.md
+265
View File
@@ -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 <name>`: commands by user.
- `/stats cmd <name>`: 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 <name>
db.stats.createIndex({ uid: 1, n: -1 }) // /stats user <name>
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
<cmd> anonymous/no-public-username bucket
<cmd>:<uid> 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?