mirror of
https://github.com/tiennm99/miti99bot.git
synced 2026-09-03 02:20:16 +00:00
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.
This commit is contained in:
+122
-77
@@ -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},
|
||||
}
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user