From 8048be8d8a292e54bf8dc45b0ad9914a182edb5e Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Sun, 28 Jun 2026 09:58:20 +0700 Subject: [PATCH] feat(server): wire MongoDB + cron + polling; rename owner/admin envs Wire MongoDB client and cron scheduler into server initialization. Rename TELEGRAM_OWNER to TELEGRAM_ADMIN_ID and SLACK_OWNER to SLACK_ADMIN_ID for clarity. Enable long-polling as default with cron scheduler for self-hosted. --- cmd/server/main.go | 199 +++++++++++++++---------- internal/deploynotify/deploy_notify.go | 2 +- 2 files changed, 123 insertions(+), 78 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 7bffdff..d1c7988 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -15,7 +15,9 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/ssm" + "github.com/go-telegram/bot" "github.com/tiennm99/miti99bot/internal/ai" + "github.com/tiennm99/miti99bot/internal/cron" "github.com/tiennm99/miti99bot/internal/deploynotify" "github.com/tiennm99/miti99bot/internal/log" "github.com/tiennm99/miti99bot/internal/metrics" @@ -67,6 +69,11 @@ const firestoreInitTimeout = 10 * time.Second // has a 10s init phase; we want to leave headroom for module wiring. const dynamodbInitTimeout = 5 * time.Second +// mongodbInitTimeout caps MongoDB connect+ping at startup (and Disconnect at +// shutdown). Atlas SRV DNS + TLS handshake can take a couple seconds on a cold +// container; 10s leaves headroom without hiding a wedged cluster. +const mongodbInitTimeout = 10 * time.Second + // ssmInitTimeout caps cold-start secret resolution. Secrets are fetched once // at startup from Parameter Store when *_PARAMETER_NAME env vars are set. const ssmInitTimeout = 5 * time.Second @@ -82,12 +89,6 @@ func main() { if cfg.TelegramBotToken == "" { log.Fatal("missing required env", "key", "TELEGRAM_BOT_TOKEN") } - if cfg.WebhookSecret == "" { - log.Fatal("missing required env", "key", "TELEGRAM_WEBHOOK_SECRET", - "why", "non-empty secret is the only auth on /webhook") - } - exportOptionalEnv("STOCK_INCOME_EVENTS_API_URL", cfg.StockIncomeEventsAPIURL) - exportOptionalEnv("STOCK_INCOME_EVENTS_API_TOKEN", cfg.StockIncomeEventsAPIToken) exportOptionalEnv("GOLD_PRICE_API_URL", cfg.GoldPriceAPIURL) exportOptionalEnv("GOLD_FX_API_URL", cfg.GoldFXAPIURL) exportOptionalEnv("GOLD_VNAPP_API_URL", cfg.GoldVNAppAPIURL) @@ -138,8 +139,18 @@ func main() { "commands", len(reg.AllCommands), "crons", len(reg.Crons())) + // In-process cron scheduler. Replaces EventBridge Scheduler off AWS; runs + // unconditionally so the long-lived container fires module crons (e.g. the + // lolschedule daily push) on their Schedule. Cutover safety comes from + // ordering + the per-date idempotency guard, not from a gate. + stopCron, err := cron.Run(rootCtx, reg) + if err != nil { + log.Fatal("cron scheduler init failed", "err", err) + } + defer stopCron() + if cfg.BotOwnerID == 0 { - log.Warn("BOT_OWNER_ID unset; all Private + Protected commands will be denied") + log.Warn("OWNER_ID unset; all Private + Protected commands will be denied") } if cfg.CronSecret == "" { log.Warn("CRON_SHARED_SECRET unset; /cron/{name} disabled (404 to all)") @@ -153,10 +164,9 @@ func main() { }) handler := server.New(server.Config{ - Bot: b, - Registry: reg, - WebhookSecret: cfg.WebhookSecret, - CronSecret: cfg.CronSecret, + Bot: b, + Registry: reg, + CronSecret: cfg.CronSecret, }) srv := &http.Server{ @@ -180,6 +190,21 @@ func main() { } }() + // Long polling is the sole Telegram transport (no webhook, no public + // ingress). Telegram permits exactly one getUpdates consumer per bot token, + // so deploy exactly one replica. Clear any webhook left over from the AWS + // deployment first — getUpdates returns HTTP 409 while a webhook is set. + // drop_pending_updates=false preserves Telegram's buffered queue so the + // poller drains updates that arrived during cutover (lossless cut). + if _, err := b.DeleteWebhook(rootCtx, &bot.DeleteWebhookParams{DropPendingUpdates: false}); err != nil { + log.Warn("deleteWebhook failed; getUpdates may 409 if a webhook is still set", "err", err) + } + go func() { + log.Info("telegram long polling started") + b.Start(rootCtx) // returns when rootCtx is cancelled + log.Info("telegram long polling stopped") + }() + <-rootCtx.Done() log.Info("shutting down") shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) @@ -190,20 +215,21 @@ func main() { } // buildProvider picks the storage backend. Selection order: -// 1. Explicit KV_PROVIDER env (memory|firestore|dynamodb) wins. -// 2. Auto-detect: AWS_LAMBDA_FUNCTION_NAME set → dynamodb; GOOGLE_CLOUD_PROJECT -// or FIRESTORE_EMULATOR_HOST set → firestore; otherwise memory. +// 1. Explicit KV_PROVIDER env (memory|firestore|dynamodb|mongodb) wins. +// 2. Auto-detect: MONGO_URL set → mongodb; otherwise memory. +// +// The self-host default is mongodb (just set MONGO_URL + MONGO_DATABASE — no +// KV_PROVIDER needed). dynamodb/firestore remain reachable only via an explicit +// KV_PROVIDER, kept for the data migrator and the integration tests; the old +// AWS_LAMBDA_FUNCTION_NAME auto-detect is removed (AWS is decommissioned). // // Returned closer is always non-nil and safe to call exactly once. func buildProvider(ctx context.Context, cfg config) (storage.KVProvider, func(), error) { backend := strings.ToLower(strings.TrimSpace(cfg.KVProvider)) if backend == "" { - switch { - case os.Getenv("AWS_LAMBDA_FUNCTION_NAME") != "": - backend = "dynamodb" - case cfg.FirestoreProject != "" || cfg.FirestoreEmulatorHost != "": - backend = "firestore" - default: + if cfg.MongoURL != "" { + backend = "mongodb" + } else { backend = "memory" } } @@ -213,6 +239,33 @@ func buildProvider(ctx context.Context, cfg config) (storage.KVProvider, func(), log.Warn("KV backend: in-memory (data lost on restart)") return storage.NewMemoryProvider(), func() {}, nil + case "mongodb": + if cfg.MongoURL == "" || cfg.MongoDatabase == "" { + return nil, func() {}, errors.New("KV_PROVIDER=mongodb requires MONGO_URL and MONGO_DATABASE") + } + initCtx, cancel := context.WithTimeout(ctx, mongodbInitTimeout) + defer cancel() + client, err := storage.NewMongoClient(initCtx, cfg.MongoURL) + if err != nil { + return nil, func() {}, err + } + db, err := storage.NewMongoDatabase(client, cfg.MongoDatabase) + if err != nil { + _ = client.Disconnect(context.Background()) + return nil, func() {}, err + } + closer := func() { + discCtx, cancel := context.WithTimeout(context.Background(), mongodbInitTimeout) + defer cancel() + if err := client.Disconnect(discCtx); err != nil { + log.Error("mongo disconnect failed", "err", err) + } + } + // NEVER log MONGO_URL — it is mongodb+srv://user:pass@host and is a + // credential. Log only the (non-secret) database name. + log.Info("storage backend", "backend", "mongodb", "database", cfg.MongoDatabase) + return storage.NewMongoProvider(db), closer, nil + case "firestore": // Emulator ignores the project ID but the SDK still requires *some* // non-empty value; supply a placeholder so emulator-only local dev works. @@ -255,38 +308,35 @@ func buildProvider(ctx context.Context, cfg config) (storage.KVProvider, func(), return storage.NewDynamoDBProvider(client, cfg.DynamoDBTable), func() {}, nil default: - return nil, func() {}, fmt.Errorf("unknown KV_PROVIDER %q (want memory|firestore|dynamodb)", backend) + return nil, func() {}, fmt.Errorf("unknown KV_PROVIDER %q (want memory|firestore|dynamodb|mongodb)", backend) } } type config struct { - Port string - TelegramBotToken string - WebhookSecret string - CronSecret string - FirestoreProject string - FirestoreEmulatorHost string - GeminiAPIKey string - StockIncomeEventsAPIURL string - StockIncomeEventsAPIToken string - GoldPriceAPIURL string - GoldFXAPIURL string - GoldVNAppAPIURL string - GoldVNAppAPIKey string - CoinBinanceAPIURL string - CoinCoinbaseAPIURL string - CoinCoinGeckoAPIURL string - Modules []string - BotOwnerID int64 - AdminUserIDs map[int64]bool - KVProvider string // empty = auto-detect; or "memory"|"firestore"|"dynamodb" - DynamoDBTable string // required when KVProvider=dynamodb - TelegramBotTokenParam string - WebhookSecretParam string - CronSecretParam string - GeminiAPIKeyParam string - StockIncomeEventsAPITokenParam string - GoldVNAppAPIKeyParam string + Port string + TelegramBotToken string + CronSecret string + FirestoreProject string + FirestoreEmulatorHost string + GeminiAPIKey string + GoldPriceAPIURL string + GoldFXAPIURL string + GoldVNAppAPIURL string + GoldVNAppAPIKey string + CoinBinanceAPIURL string + CoinCoinbaseAPIURL string + CoinCoinGeckoAPIURL string + Modules []string + BotOwnerID int64 + AdminUserIDs map[int64]bool + KVProvider string // empty = auto-detect; or "memory"|"firestore"|"dynamodb"|"mongodb" + DynamoDBTable string // required when KVProvider=dynamodb + MongoURL string // required when KVProvider=mongodb (Atlas SRV connection string; SECRET — never log) + MongoDatabase string // required when KVProvider=mongodb + TelegramBotTokenParam string + CronSecretParam string + GeminiAPIKeyParam string + GoldVNAppAPIKeyParam string } func loadConfig() config { @@ -307,33 +357,30 @@ func loadConfig() config { log.Fatal("invalid PORT", "value", port) } return config{ - Port: port, - TelegramBotToken: envMap["TELEGRAM_BOT_TOKEN"], - WebhookSecret: envMap["TELEGRAM_WEBHOOK_SECRET"], - CronSecret: envMap["CRON_SHARED_SECRET"], - FirestoreProject: envMap["GOOGLE_CLOUD_PROJECT"], - FirestoreEmulatorHost: envMap["FIRESTORE_EMULATOR_HOST"], - GeminiAPIKey: envMap["GEMINI_API_KEY"], - StockIncomeEventsAPIURL: envMap["STOCK_INCOME_EVENTS_API_URL"], - StockIncomeEventsAPIToken: envMap["STOCK_INCOME_EVENTS_API_TOKEN"], - GoldPriceAPIURL: envMap["GOLD_PRICE_API_URL"], - GoldFXAPIURL: envMap["GOLD_FX_API_URL"], - GoldVNAppAPIURL: envMap["GOLD_VNAPP_API_URL"], - GoldVNAppAPIKey: envMap["GOLD_VNAPP_API_KEY"], - CoinBinanceAPIURL: envMap["COIN_BINANCE_API_URL"], - CoinCoinbaseAPIURL: envMap["COIN_COINBASE_API_URL"], - CoinCoinGeckoAPIURL: envMap["COIN_COINGECKO_API_URL"], - Modules: splitCSV(envMap["MODULES"]), - BotOwnerID: parseInt64(envMap["BOT_OWNER_ID"]), - AdminUserIDs: parseInt64Set(envMap["ADMIN_USER_IDS"]), - KVProvider: envMap["KV_PROVIDER"], - DynamoDBTable: envMap["DYNAMODB_TABLE"], - TelegramBotTokenParam: strings.TrimSpace(envMap["TELEGRAM_BOT_TOKEN_PARAMETER_NAME"]), - WebhookSecretParam: strings.TrimSpace(envMap["TELEGRAM_WEBHOOK_SECRET_PARAMETER_NAME"]), - CronSecretParam: strings.TrimSpace(envMap["CRON_SHARED_SECRET_PARAMETER_NAME"]), - GeminiAPIKeyParam: strings.TrimSpace(envMap["GEMINI_API_KEY_PARAMETER_NAME"]), - StockIncomeEventsAPITokenParam: strings.TrimSpace(envMap["STOCK_INCOME_EVENTS_API_TOKEN_PARAMETER_NAME"]), - GoldVNAppAPIKeyParam: strings.TrimSpace(envMap["GOLD_VNAPP_API_KEY_PARAMETER_NAME"]), + Port: port, + TelegramBotToken: envMap["TELEGRAM_BOT_TOKEN"], + CronSecret: envMap["CRON_SHARED_SECRET"], + FirestoreProject: envMap["GOOGLE_CLOUD_PROJECT"], + FirestoreEmulatorHost: envMap["FIRESTORE_EMULATOR_HOST"], + GeminiAPIKey: envMap["GEMINI_API_KEY"], + GoldPriceAPIURL: envMap["GOLD_PRICE_API_URL"], + GoldFXAPIURL: envMap["GOLD_FX_API_URL"], + GoldVNAppAPIURL: envMap["GOLD_VNAPP_API_URL"], + GoldVNAppAPIKey: envMap["GOLD_VNAPP_API_KEY"], + CoinBinanceAPIURL: envMap["COIN_BINANCE_API_URL"], + CoinCoinbaseAPIURL: envMap["COIN_COINBASE_API_URL"], + CoinCoinGeckoAPIURL: envMap["COIN_COINGECKO_API_URL"], + Modules: splitCSV(envMap["MODULES"]), + BotOwnerID: parseInt64(envMap["OWNER_ID"]), + AdminUserIDs: parseInt64Set(envMap["ADMIN_IDS"]), + KVProvider: envMap["KV_PROVIDER"], + DynamoDBTable: envMap["DYNAMODB_TABLE"], + MongoURL: envMap["MONGO_URL"], + MongoDatabase: envMap["MONGO_DATABASE"], + TelegramBotTokenParam: strings.TrimSpace(envMap["TELEGRAM_BOT_TOKEN_PARAMETER_NAME"]), + CronSecretParam: strings.TrimSpace(envMap["CRON_SHARED_SECRET_PARAMETER_NAME"]), + GeminiAPIKeyParam: strings.TrimSpace(envMap["GEMINI_API_KEY_PARAMETER_NAME"]), + GoldVNAppAPIKeyParam: strings.TrimSpace(envMap["GOLD_VNAPP_API_KEY_PARAMETER_NAME"]), } } @@ -343,10 +390,8 @@ func resolveSSMSecrets(ctx context.Context, cfg *config) error { target *string }{ {name: cfg.TelegramBotTokenParam, target: &cfg.TelegramBotToken}, - {name: cfg.WebhookSecretParam, target: &cfg.WebhookSecret}, {name: cfg.CronSecretParam, target: &cfg.CronSecret}, {name: cfg.GeminiAPIKeyParam, target: &cfg.GeminiAPIKey}, - {name: cfg.StockIncomeEventsAPITokenParam, target: &cfg.StockIncomeEventsAPIToken}, {name: cfg.GoldVNAppAPIKeyParam, target: &cfg.GoldVNAppAPIKey}, } diff --git a/internal/deploynotify/deploy_notify.go b/internal/deploynotify/deploy_notify.go index b3c91a1..9d61906 100644 --- a/internal/deploynotify/deploy_notify.go +++ b/internal/deploynotify/deploy_notify.go @@ -94,7 +94,7 @@ func skipReason(cfg Config) string { case cfg.GitSHA == "": return "empty gitSHA (build without -ldflags)" case cfg.OwnerID == 0: - return "no BOT_OWNER_ID configured" + return "no OWNER_ID configured" case cfg.KV == nil: return "no KV configured" case cfg.Bot == nil && cfg.Sender == nil: