scrabble-game/backend/cmd/backend/main.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/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()
}