2207ac6132
Add an in-memory SendLimiter enforcing a one-per-minute cooldown and a five-per-rolling-hour cap per recipient address, checked before provisioning or sending in RequestCode, RequestLoginCode and RequestLinkCode. It guards against email bombing and protects the relay quota. The limiter is injected in main (nil in tests, so the domain suite is unaffected); ErrTooManyRequests maps to HTTP 429.
307 lines
12 KiB
Go
307 lines
12 KiB
Go
// 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/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/link"
|
|
"scrabble/backend/internal/lobby"
|
|
"scrabble/backend/internal/notify"
|
|
"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
|
|
|
|
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")
|
|
|
|
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("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 persisted active dictionary version with the registry: a
|
|
// version activated through the admin console (and written to the dictionary
|
|
// volume) is adopted again after a restart; otherwise the configured seed
|
|
// version is kept and persisted (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"))
|
|
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))
|
|
|
|
// 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.
|
|
links := link.NewService(emails, accounts, accountmerge.NewMerger(db), 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)
|
|
|
|
// 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))
|
|
|
|
// 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,
|
|
Notifier: hub,
|
|
ExportSignKey: cfg.ExportSignKey,
|
|
Renderer: renderer,
|
|
})
|
|
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))
|
|
errc := make(chan error, 2)
|
|
go func() { errc <- pushSrv.Run(ctx) }()
|
|
go func() { errc <- srv.Run(ctx) }()
|
|
err = <-errc
|
|
cancel()
|
|
<-errc
|
|
return 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()
|
|
}
|