// Command backend is the Scrabble platform's internal domain service. It boots // the OpenTelemetry runtime, opens the Postgres pool and applies migrations, // loads the dictionaries into the engine registry, warms the session cache, // constructs the game domain and starts its turn-timeout sweeper, constructs the // lobby and social domains, then serves the HTTP listener with the infrastructure // probes and the /api/v1 route group, behind which the domains expose their HTTP // endpoints to the gateway. package main import ( "context" "fmt" "log" "os/signal" "syscall" "time" "github.com/google/uuid" "go.uber.org/zap" "scrabble/backend/internal/account" "scrabble/backend/internal/accountmerge" "scrabble/backend/internal/adminalert" "scrabble/backend/internal/ads" "scrabble/backend/internal/banview" "scrabble/backend/internal/config" "scrabble/backend/internal/connector" "scrabble/backend/internal/engine" "scrabble/backend/internal/feedback" "scrabble/backend/internal/game" "scrabble/backend/internal/gamelimits" "scrabble/backend/internal/link" "scrabble/backend/internal/lobby" "scrabble/backend/internal/notify" "scrabble/backend/internal/payments" "scrabble/backend/internal/postgres" "scrabble/backend/internal/pushgrpc" "scrabble/backend/internal/ratewatch" "scrabble/backend/internal/render" "scrabble/backend/internal/robot" "scrabble/backend/internal/server" "scrabble/backend/internal/session" "scrabble/backend/internal/social" "scrabble/backend/internal/telemetry" ) // telemetryShutdownTimeout bounds the OpenTelemetry flush during process exit. const telemetryShutdownTimeout = 5 * time.Second // adminAlertInterval is how often the operator-alert worker checks for new feedback / // complaints; a burst within one interval coalesces into a single digest email. const adminAlertInterval = 5 * time.Minute func main() { cfg, err := config.Load() if err != nil { log.Fatalf("backend: load config: %v", err) } logger, err := newLogger(cfg.LogLevel) if err != nil { log.Fatalf("backend: build logger: %v", err) } defer func() { _ = logger.Sync() }() ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() if err := run(ctx, cfg, logger); err != nil { logger.Fatal("backend: terminated", zap.Error(err)) } } // run wires the process dependencies in order — telemetry, database (with // migrations), engine dictionaries, session cache, game domain (with its // turn-timeout sweeper), the robot opponent (pool + move driver) and the // matchmaking reaper, HTTP server — and blocks until ctx is cancelled. func run(ctx context.Context, cfg config.Config, logger *zap.Logger) error { // A cancellable child context so the first server (or signal) to stop tears // the rest down — the HTTP and gRPC listeners and every background worker // share it. ctx, cancel := context.WithCancel(ctx) defer cancel() tel, err := telemetry.New(ctx, cfg.Telemetry) if err != nil { return fmt.Errorf("init telemetry: %w", err) } defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), telemetryShutdownTimeout) defer cancel() if err := tel.Shutdown(shutdownCtx); err != nil { logger.Warn("telemetry shutdown", zap.Error(err)) } }() if err := tel.StartRuntimeMetrics(); err != nil { logger.Warn("telemetry: start runtime metrics", zap.Error(err)) } db, err := postgres.Open(ctx, cfg.Postgres, postgres.WithTracerProvider(tel.TracerProvider()), postgres.WithMeterProvider(tel.MeterProvider()), ) if err != nil { return fmt.Errorf("open database: %w", err) } defer func() { _ = db.Close() }() if err := postgres.ApplyMigrations(ctx, db); err != nil { return fmt.Errorf("apply migrations: %w", err) } logger.Info("database migrations applied") // Deliver the build's dictionary version onto the persistent volume if a redeploy // bumped it (add-only; old versions in-flight games pin are untouched), so the new // dictionary is resident and can be activated below without an admin upload. if err := engine.DeliverVersion(cfg.Game.DictDir, cfg.Game.DictSeedDir, cfg.Game.DictVersion); err != nil { return fmt.Errorf("deliver dictionary version: %w", err) } registry, err := engine.OpenWithVersions(cfg.Game.DictDir, cfg.Game.DictVersion) if err != nil { return fmt.Errorf("load dictionaries: %w", err) } defer func() { _ = registry.Close() }() logger.Info("dictionaries loaded", zap.String("dir", cfg.Game.DictDir), zap.String("seed_dir", cfg.Game.DictSeedDir), zap.String("version", cfg.Game.DictVersion)) // Admin console: an optional backend client to the Telegram connector // side-service for operator broadcasts. Unset (BACKEND_CONNECTOR_ADDR empty) // leaves broadcasts disabled — the console shows a "not configured" notice. var conn *connector.Client if cfg.ConnectorAddr != "" { conn, err = connector.New(cfg.ConnectorAddr) if err != nil { return fmt.Errorf("dial connector: %w", err) } defer func() { _ = conn.Close() }() logger.Info("connector client ready", zap.String("addr", cfg.ConnectorAddr)) } sessions := session.NewService(session.NewStore(db), session.NewCache()) if err := sessions.Warm(ctx); err != nil { return fmt.Errorf("warm session cache: %w", err) } logger.Info("session cache warmed") // The in-process live-event hub fans domain intents out to the gRPC push // stream. It is installed on every emitting service before any background // worker starts so robot moves and timeout sweeps also emit. hub := notify.NewHub(0) accounts := account.NewStore(db) accounts.SetMetrics(tel.MeterProvider().Meter("scrabble/backend/account")) games := game.NewService(game.NewStore(db), accounts, registry, cfg.Game, logger) // Reconcile the active dictionary version with the registry: the build version // (delivered above) becomes active so a redeploy that bumped it goes live for new // games, except a newer version installed out-of-band through the admin console, // which is not downgraded by a restart (docs/ARCHITECTURE.md §5). if err := games.InitActiveVersion(ctx); err != nil { return fmt.Errorf("init active dictionary version: %w", err) } logger.Info("active dictionary version", zap.String("version", games.ActiveVersion())) games.SetNotifier(hub) games.SetMetrics(tel.MeterProvider().Meter("scrabble/backend/game")) // Active-game limit config: the per-tier, per-kind caps in backend.config, read once into an // in-memory cache at boot and refreshed when the admin edits them. A boot-time load fails fast if // the single config row is missing; the game domain reads the cache on every new-game gate. gameLimits := gamelimits.NewService(gamelimits.NewStore(db)) if err := gameLimits.Load(ctx); err != nil { return fmt.Errorf("load game-limit config: %w", err) } games.SetGameLimits(gameLimits) logger.Info("game-limit config loaded") go games.RunSweeper(ctx, cfg.Game.TimeoutSweepInterval) logger.Info("game turn-timeout sweeper started", zap.Duration("interval", cfg.Game.TimeoutSweepInterval)) // Reap abandoned guest accounts (no game seat, account age past // the retention window). Dependent rows fall away via ON DELETE CASCADE. guestReaper := account.NewGuestReaper(accounts, cfg.GuestRetention, logger) go guestReaper.Run(ctx, cfg.GuestReapInterval) logger.Info("guest reaper started", zap.Duration("interval", cfg.GuestReapInterval), zap.Duration("retention", cfg.GuestRetention)) // Purge the account-deletion legal dossier past its retention TTL: the // retained-identities journal, and the feedback thread + dossier PII of long-deleted // accounts (chat is kept). Checked daily; the TTL is a two-year policy constant. retentionReaper := account.NewRetentionReaper(accounts, account.RetentionTTL, logger) go retentionReaper.Run(ctx, 24*time.Hour) logger.Info("retention reaper started", zap.Duration("interval", 24*time.Hour), zap.Duration("retention", account.RetentionTTL)) // Re-evaluate moderated-chat write access when a temporary block self-expires: // no operator action fires then, so the sweeper emits the chat-access-changed // event for lapsed blocks and the gateway re-pushes the chat-gate command. chatSweeper := account.NewSuspensionSweeper(accounts, func(id uuid.UUID) { hub.Publish(notify.ChatAccessChanged(id)) }, logger) go chatSweeper.Run(ctx) logger.Info("suspension expiry sweeper started", zap.Duration("interval", chatSweeper.Interval())) // Lobby & social domains. Their REST and stream surface lives in the gateway, // so they are handed to the server (like the route groups) for the handlers. mailer := newMailer(cfg.SMTP, logger) emails := account.NewEmailService(accounts, mailer, cfg.PublicBaseURL) // Throttle confirm-code sends per recipient: at most one per minute and five per // rolling hour, guarding against email bombing and the relay's own quota. emails.SetSendLimiter(account.NewSendLimiter(time.Minute, 5)) // Account linking & merge: the orchestrator over the account, merge and // session layers. Wired to the /api/v1/user/link REST surface below. merger := accountmerge.NewMerger(db) links := link.NewService(emails, accounts, merger, sessions) socialSvc := social.NewService(social.NewStore(db), accounts, games) socialSvc.SetNotifier(hub) socialSvc.SetMetrics(tel.MeterProvider().Meter("scrabble/backend/social")) // A nudge the recipient answered by moving is marked read on the move path; every nudge in a // game is marked read when the game finishes (a stale badge), on any completion path. games.SetNudgeClearer(socialSvc.ClearNudges) games.SetNudgeExpirer(socialSvc.ExpireNudges) // Reap per-game disguised-robot friend requests once their game is long finished // (the robot ignores them; the row only pins the in-game "request sent" state). robotReqReaper := social.NewRobotFriendRequestReaper(socialSvc, logger) go robotReqReaper.Run(ctx) logger.Info("robot friend request reaper started", zap.Duration("interval", robotReqReaper.Interval()), zap.Duration("retention", robotReqReaper.Retention())) feedbackSvc := feedback.NewService(feedback.NewStore(db), accounts) feedbackSvc.SetNotifier(hub) // Operator alert emails on new feedback / word complaints, coalesced into one digest // per interval. Inert unless a distinct admin sender and recipient are configured. if cfg.SMTP.AdminFrom != "" && cfg.SMTP.AdminTo != "" { // The alert digest carries no admin-console link on purpose — an admin URL must not // travel in an email (a mail provider could cache or index it); see adminalert.New. alerts := adminalert.New(mailer, feedbackSvc, games, cfg.SMTP.AdminFrom, cfg.SMTP.AdminTo, logger) go alerts.Run(ctx, adminAlertInterval) logger.Info("admin alert worker started", zap.Duration("interval", adminAlertInterval)) } // Robot opponent: provision its durable account pool (a hard startup // dependency, like the dictionaries) and start its move driver. The matchmaker // substitutes a pooled robot for a missing human after the wait window. robots := robot.NewService(games, accounts, socialSvc, tel.MeterProvider().Meter("scrabble/backend/robot"), logger) if err := robots.EnsurePool(ctx); err != nil { return fmt.Errorf("provision robot pool: %w", err) } // Honest-AI fast path: a move in a vs_ai game triggers the robot's reply at once // (the periodic driver below is the fallback). Set after the pool is provisioned. games.SetAITrigger(robots.TriggerMove) go robots.Run(ctx, cfg.Robot.DriveInterval) logger.Info("robot driver started", zap.Duration("interval", cfg.Robot.DriveInterval)) matchmaker := lobby.NewMatchmaker(games, robots, cfg.Lobby.RobotWait, cfg.Lobby.RobotWaitJitter, logger) matchmaker.SetNotifier(hub) matchmaker.SetBlocker(socialSvc) go matchmaker.RunReaper(ctx, cfg.Lobby.ReaperInterval) invitations := lobby.NewInvitationService(lobby.NewStore(db), games, accounts, socialSvc) invitations.SetNotifier(hub) logger.Info("lobby and social domains ready", zap.Duration("robot_wait", cfg.Lobby.RobotWait), zap.Duration("robot_wait_jitter", cfg.Lobby.RobotWaitJitter)) // Rate-limit observability: ingest the gateway's rejection reports for the // admin throttled view and the conservative high-rate auto-flag. rateWatch := ratewatch.New(cfg.RateWatch, accounts, logger) logger.Info("rate watch ready", zap.Int("flag_threshold", cfg.RateWatch.FlagThreshold), zap.Duration("flag_window", cfg.RateWatch.FlagWindow)) // Ban observability: mirror the gateway's active IP bans for the admin console's // active-bans panel and collect operator unban requests. banView := banview.New() // Advertising-banner domain: campaign rotation feeding the profile.get banner // block and the banner admin console section. adsSvc := ads.NewService(ads.NewStore(db)) // In-game currency domain (data foundation): the payments schema behind a // narrow interface. A boot-time reachability check fails fast if the schema // did not migrate; the wallet routes are registered when that surface lands. paymentsSvc := payments.NewService(payments.NewStore(db)) if err := paymentsSvc.Ping(ctx); err != nil { return fmt.Errorf("payments schema unreachable: %w", err) } logger.Info("payments domain ready") // Warm the public-offer price list cache so /offer/ serves the current catalog from the first // request; it is reprojected lazily thereafter on any catalog edit. Non-fatal — a transient // failure here only defers the projection to the first read. if _, err := paymentsSvc.OfferPricing(ctx); err != nil { logger.Warn("offer pricing warm failed; will project on first request", zap.Error(err)) } // Wire the payments surface into the domains that consume it: the online-game hint wallet // and the account-merge wallet fold. Done after the reachability check so a broken payments // schema fails boot before anything depends on it. games.SetHintWallet(paymentsSvc) merger.SetPayments(paymentsSvc) // The image-render sidecar client for the PNG export artifact; nil (PNG // download answers 404) when BACKEND_RENDERER_URL is unset. var renderer *render.Client if cfg.RendererURL != "" { renderer = render.New(cfg.RendererURL) } srv := server.New(cfg.HTTPAddr, server.Deps{ Logger: logger, DB: db, PingTimeout: cfg.Postgres.OperationTimeout, SessionsReady: sessions.Ready, Sessions: sessions, Accounts: accounts, Games: games, Feedback: feedbackSvc, Social: socialSvc, Matchmaker: matchmaker, Invitations: invitations, Emails: emails, Links: links, Registry: registry, DictDir: cfg.Game.DictDir, Connector: conn, RateWatch: rateWatch, BanView: banView, Ads: adsSvc, Payments: paymentsSvc, GameLimits: gameLimits, Notifier: hub, ExportSignKey: cfg.ExportSignKey, Renderer: renderer, Robokassa: cfg.Robokassa, }) pushSrv := pushgrpc.NewServer(cfg.GRPCAddr, hub, logger) // Run the HTTP and gRPC push listeners together; the first to stop (a listen // error, or ctx cancellation on signal) tears down the other through cancel. logger.Info("servers starting", zap.String("http_addr", cfg.HTTPAddr), zap.String("grpc_addr", cfg.GRPCAddr)) // Sweep expired pending payment orders on a cadence (cosmetic hygiene; a late valid callback // still credits). Runs until ctx is cancelled. go runOrderReaper(ctx, paymentsSvc, logger) // Deliver pending payment_events to connected clients as an in-app wallet-refresh push (the // credit already landed in the ledger; a return-focus poll is the client-side fallback). go runPaymentDispatcher(ctx, paymentsSvc, hub, logger) errc := make(chan error, 2) go func() { errc <- pushSrv.Run(ctx) }() go func() { errc <- srv.Run(ctx) }() err = <-errc cancel() <-errc return err } // runOrderReaper periodically expires pending payment orders past their configured lifetime, until // ctx is cancelled. Expiry is cosmetic: a later valid provider callback still credits an expired // order. func runOrderReaper(ctx context.Context, p *payments.Service, log *zap.Logger) { t := time.NewTicker(5 * time.Minute) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: if n, err := p.ExpireOrders(ctx); err != nil { log.Warn("order reaper: sweep failed", zap.Error(err)) } else if n > 0 { log.Info("order reaper: expired pending orders", zap.Int("count", n)) } } } } // runPaymentDispatcher delivers pending payment_events to connected clients as a wallet-refresh // signal (KindNotification / "payment"), marking each delivered, until ctx is cancelled. The credit // already committed to the ledger; this is only the in-app push so an open wallet updates in place. func runPaymentDispatcher(ctx context.Context, p *payments.Service, pub notify.Publisher, log *zap.Logger) { t := time.NewTicker(3 * time.Second) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: evs, err := p.UndispatchedEvents(ctx, 50) if err != nil { log.Warn("payment dispatcher: read failed", zap.Error(err)) continue } for _, e := range evs { pub.Publish(notify.Notification(e.AccountID, notify.NotifyPayment)) if err := p.MarkEventDispatched(ctx, e.EventID); err != nil { log.Warn("payment dispatcher: mark failed", zap.String("event", e.EventID.String()), zap.Error(err)) } } } } } // newMailer builds the confirm-code mailer: an SMTP relay when a host is // configured, otherwise the development log mailer (the code is logged, not sent). func newMailer(cfg account.SMTPConfig, logger *zap.Logger) account.Mailer { if cfg.Host == "" { logger.Info("email: using log mailer (BACKEND_SMTP_HOST unset)") return account.NewLogMailer(logger) } logger.Info("email: using SMTP relay", zap.String("host", cfg.Host)) return account.NewSMTPMailer(cfg) } // newLogger builds a production JSON logger at the given level. func newLogger(level string) (*zap.Logger, error) { var lvl zap.AtomicLevel if err := lvl.UnmarshalText([]byte(level)); err != nil { return nil, err } cfg := zap.NewProductionConfig() cfg.Level = lvl return cfg.Build() }