scrabble-game/gateway/cmd/gateway/main.go

// Command gateway is the Scrabble platform's only public ingress. It terminates
// the client's Connect-RPC/FlatBuffers traffic over h2c, validates platform /
// email / guest credentials and mints opaque sessions, rate-limits, injects
// X-User-ID when forwarding to the backend over REST, and bridges the backend's
// gRPC push stream to each client's in-app live channel. It also serves the
// backend's admin console at /_gm on the public listener behind HTTP Basic-Auth.
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net"
	"net/http"
	"os/signal"
	"syscall"
	"time"

	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"

	"scrabble/gateway/internal/admin"
	"scrabble/gateway/internal/backendclient"
	"scrabble/gateway/internal/botlink"
	"scrabble/gateway/internal/config"
	"scrabble/gateway/internal/connector"
	"scrabble/gateway/internal/connectsrv"
	"scrabble/gateway/internal/push"
	"scrabble/gateway/internal/ratelimit"
	"scrabble/gateway/internal/session"
	"scrabble/gateway/internal/transcode"
	"scrabble/pkg/mtls"
	botlinkv1 "scrabble/pkg/proto/botlink/v1"
	telegramv1 "scrabble/pkg/proto/telegram/v1"
	pkgtel "scrabble/pkg/telemetry"
)

const (
	// botLinkKeepaliveTime is how often the gateway pings an idle bot-link stream
	// to hold the WAN connection open and detect a dead bot.
	botLinkKeepaliveTime = 30 * time.Second
	// botLinkKeepaliveTimeout bounds the wait for a keepalive ping reply.
	botLinkKeepaliveTimeout = 10 * time.Second
	// botLinkMinPingInterval is the smallest client ping interval the gateway
	// tolerates before treating it as abuse.
	botLinkMinPingInterval = 10 * time.Second
)

const (
	// shutdownTimeout bounds the graceful HTTP shutdown.
	shutdownTimeout = 10 * time.Second
	// telemetryShutdownTimeout bounds the OpenTelemetry flush during process exit.
	telemetryShutdownTimeout = 5 * time.Second
	// pushReconnectDelay is the pause before re-subscribing to the backend push
	// stream after it ends.
	pushReconnectDelay = 2 * time.Second
	// gatewayID identifies this gateway instance to the backend push channel.
	gatewayID = "gateway"
	// readHeaderTimeout bounds reading one request's headers on the public
	// listener (a slowloris guard). Bodies and long-lived streams are governed by
	// the h2c settings in connectsrv — Read/WriteTimeout stay unset on purpose,
	// they would kill the Subscribe stream.
	readHeaderTimeout = 10 * time.Second
	// throttleReportInterval is the cadence of the rate-limiter rejection
	// summary: the Warn log per throttled key and the report to the backend.
	throttleReportInterval = 30 * time.Second
	// banSyncInterval is the cadence of the active-ban sync to the backend (which
	// feeds the admin-console view) and the operator-unban pull.
	banSyncInterval = 30 * time.Second
)

func main() {
	cfg, err := config.Load()
	if err != nil {
		log.Fatalf("gateway: load config: %v", err)
	}
	logger, err := newLogger(cfg.LogLevel)
	if err != nil {
		log.Fatalf("gateway: 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("gateway: terminated", zap.Error(err))
	}
}

// run wires the gateway dependencies and serves the public listener (which also
// fronts the admin console at /_gm) until the context is cancelled.
func run(ctx context.Context, cfg config.Config, logger *zap.Logger) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	tel, err := pkgtel.New(ctx, cfg.Telemetry)
	if err != nil {
		return err
	}
	defer func() {
		shutdownCtx, sc := context.WithTimeout(context.Background(), telemetryShutdownTimeout)
		defer sc()
		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))
	}

	backend, err := backendclient.New(cfg.BackendHTTPURL, cfg.BackendGRPCAddr, cfg.BackendTimeout)
	if err != nil {
		return err
	}
	defer func() { _ = backend.Close() }()

	sessions := session.NewCache(backend, cfg.SessionTTL, cfg.SessionCacheMax)
	limiter := ratelimit.New()
	tracker := ratelimit.NewTracker()
	banlist := ratelimit.NewBanlist(ratelimit.BanConfig{
		Enabled:   cfg.Abuse.BanEnabled,
		Threshold: cfg.Abuse.BanThreshold,
		Window:    cfg.Abuse.BanWindow,
		Duration:  cfg.Abuse.BanDuration,
	})
	hub := push.NewHub(0)

	var validator transcode.TelegramValidator
	if cfg.ValidatorAddr != "" {
		conn, cerr := connector.New(cfg.ValidatorAddr)
		if cerr != nil {
			return cerr
		}
		defer func() { _ = conn.Close() }()
		validator = conn
	} else {
		logger.Warn("telegram auth disabled (GATEWAY_VALIDATOR_ADDR unset)")
	}

	// The reverse bot-link: the remote Telegram bot dials this gateway over mTLS and
	// the gateway pushes send commands down the stream. Out-of-app push is
	// fire-and-forget; the backend admin relay (plaintext, internal) awaits the Ack.
	var botHub *botlink.Hub
	if cfg.BotLinkEnabled() {
		botHub = botlink.NewHub(logger, tel.MeterProvider().Meter("scrabble/gateway/botlink"),
			func(ctx context.Context, externalID string) (bool, bool, error) {
				r, rerr := backend.ChatEligibility(ctx, externalID)
				return r.Registered, r.Eligible, rerr
			})
		tlsCfg, terr := mtls.ServerConfig(cfg.BotLink.CertFile, cfg.BotLink.KeyFile, cfg.BotLink.CAFile)
		if terr != nil {
			return terr
		}
		botSrv := grpc.NewServer(
			grpc.Creds(credentials.NewTLS(tlsCfg)),
			grpc.StatsHandler(otelgrpc.NewServerHandler()),
			grpc.KeepaliveParams(keepalive.ServerParameters{Time: botLinkKeepaliveTime, Timeout: botLinkKeepaliveTimeout}),
			grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{MinTime: botLinkMinPingInterval, PermitWithoutStream: true}),
		)
		botlinkv1.RegisterBotLinkServer(botSrv, botHub)
		if serr := serveGRPC(ctx, "botlink", cfg.BotLink.Addr, botSrv, logger); serr != nil {
			return serr
		}
		if cfg.BotLink.RelayAddr != "" {
			relaySrv := grpc.NewServer(grpc.StatsHandler(otelgrpc.NewServerHandler()))
			telegramv1.RegisterTelegramServer(relaySrv, botlink.NewRelayServer(botHub, cfg.BotLink.SendTimeout))
			if serr := serveGRPC(ctx, "botlink-relay", cfg.BotLink.RelayAddr, relaySrv, logger); serr != nil {
				return serr
			}
		}
	} else {
		logger.Warn("telegram bot channel disabled (GATEWAY_BOTLINK_ADDR unset)")
	}

	// The admin console (backend /_gm) is fronted on the public listener behind
	// Basic-Auth, enabled when both credentials are set; it is mounted on the edge
	// mux so the Connect h2c handler stays the top-level handler.
	var adminProxy http.Handler
	if cfg.AdminEnabled() {
		adminProxy, err = admin.NewProxy(cfg.BackendHTTPURL, cfg.AdminUser, cfg.AdminPassword, logger)
		if err != nil {
			return err
		}
	} else {
		logger.Info("admin console disabled (set GATEWAY_ADMIN_USER and GATEWAY_ADMIN_PASSWORD)")
	}

	registry := transcode.NewRegistry(backend, validator, transcode.WithVKAuth(cfg.VKAppSecret))
	edge := connectsrv.NewServer(connectsrv.Deps{
		Registry:     registry,
		Sessions:     sessions,
		Backend:      backend,
		Limiter:      limiter,
		Tracker:      tracker,
		Banlist:      banlist,
		Honeytoken:   cfg.Abuse.Honeytoken,
		Hub:          hub,
		RateLimit:    cfg.RateLimit,
		Heartbeat:    cfg.PushHeartbeatInterval,
		Logger:       logger,
		AdminProxy:   adminProxy,
		Meter:        tel.MeterProvider().Meter("scrabble/gateway/edge"),
		MaxBodyBytes: cfg.MaxBodyBytes,
	})

	// Bridge the backend push stream into the fan-out hub (and the out-of-app
	// channel via the bot-link).
	go runPushPump(ctx, backend, hub, botHub, logger)
	// Periodically summarise rate-limiter rejections (Warn log + backend report).
	go runThrottleReporter(ctx, tracker, backend, logger)
	// When the IP ban is enabled (prod), sync the active set to the backend (the
	// admin-console view) and apply the operator unbans it returns.
	if cfg.Abuse.BanEnabled {
		go runBanSync(ctx, banlist, backend, logger)
	}

	public := &http.Server{Addr: cfg.HTTPAddr, Handler: edge.HTTPHandler(), ReadHeaderTimeout: readHeaderTimeout}
	servers := []*namedServer{{name: "public", srv: public}}

	logger.Info("gateway starting",
		zap.String("http_addr", cfg.HTTPAddr),
		zap.String("backend_http", cfg.BackendHTTPURL),
		zap.String("backend_grpc", cfg.BackendGRPCAddr))
	return runServers(ctx, cancel, servers, logger)
}

// namedServer pairs an HTTP server with a label for diagnostics.
type namedServer struct {
	name string
	srv  *http.Server
}

// runServers serves every listener and shuts them all down when the first one
// stops or the context is cancelled.
func runServers(ctx context.Context, cancel context.CancelFunc, servers []*namedServer, logger *zap.Logger) error {
	errc := make(chan error, len(servers))
	for _, s := range servers {
		go func(s *namedServer) {
			logger.Info("listener starting", zap.String("server", s.name), zap.String("addr", s.srv.Addr))
			err := s.srv.ListenAndServe()
			if errors.Is(err, http.ErrServerClosed) {
				err = nil
			}
			errc <- err
		}(s)
	}

	var first error
	select {
	case <-ctx.Done():
	case first = <-errc:
	}
	cancel()

	shutdownCtx, sc := context.WithTimeout(context.Background(), shutdownTimeout)
	defer sc()
	for _, s := range servers {
		if err := s.srv.Shutdown(shutdownCtx); err != nil {
			logger.Warn("listener shutdown", zap.String("server", s.name), zap.Error(err))
		}
	}
	return first
}

// serveGRPC starts a gRPC server on addr in the background and gracefully stops it
// when ctx is cancelled. It returns synchronously on a bind error so a
// misconfigured listener fails startup fast; a later Serve error is logged.
func serveGRPC(ctx context.Context, name, addr string, srv *grpc.Server, logger *zap.Logger) error {
	lis, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("gateway: listen %s (%s): %w", addr, name, err)
	}
	go func() {
		<-ctx.Done()
		srv.GracefulStop()
	}()
	go func() {
		logger.Info("listener starting", zap.String("server", name), zap.String("addr", addr))
		if err := srv.Serve(lis); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
			logger.Error("grpc listener failed", zap.String("server", name), zap.Error(err))
		}
	}()
	return nil
}

// runThrottleReporter drains the rate-limiter rejection tracker on a fixed
// cadence, emits one Warn summary per throttled key and forwards the report to
// the backend (which feeds the admin throttled view and the high-rate
// auto-flag), until the context is done. A failed delivery is logged and
// dropped — the next window reports fresh data anyway.
func runThrottleReporter(ctx context.Context, tracker *ratelimit.Tracker, backend *backendclient.Client, logger *zap.Logger) {
	ticker := time.NewTicker(throttleReportInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		entries := tracker.Drain()
		if len(entries) == 0 {
			continue
		}
		for _, e := range entries {
			logger.Warn("rate limited",
				zap.String("class", e.Class),
				zap.String("key", e.Key),
				zap.Int("rejected", e.Rejected),
				zap.Duration("window", throttleReportInterval))
		}
		if err := backend.ReportRateLimited(ctx, int(throttleReportInterval.Seconds()), entries); err != nil {
			logger.Warn("rate-limit report failed", zap.Error(err))
		}
	}
}

// runBanSync periodically reports the gateway's active IP bans to the backend (the
// admin-console view) and applies the operator unbans it returns, until the
// context is done. A failed sync is logged and dropped — the next tick reports
// fresh state, and a missed unban is retried on it.
func runBanSync(ctx context.Context, banlist *ratelimit.Banlist, backend *backendclient.Client, logger *zap.Logger) {
	ticker := time.NewTicker(banSyncInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		unban, err := backend.SyncBans(ctx, banlist.Active())
		if err != nil {
			logger.Warn("ban sync failed", zap.Error(err))
			continue
		}
		for _, ip := range unban {
			banlist.Unban(ip)
			logger.Info("ban cleared by operator", zap.String("client_ip", ip))
		}
	}
}

// runPushPump keeps a backend push subscription open, forwarding every event to
// the hub and re-subscribing after the stream ends, until the context is done. For
// the out-of-app push kinds it also routes events whose recipient has no live
// in-app stream to the platform connector (a nil connector disables that channel).
func runPushPump(ctx context.Context, backend *backendclient.Client, hub *push.Hub, bot *botlink.Hub, logger *zap.Logger) {
	for ctx.Err() == nil {
		stream, err := backend.SubscribePush(ctx, gatewayID)
		if err != nil {
			logger.Warn("push subscribe failed", zap.Error(err))
			if !sleep(ctx, pushReconnectDelay) {
				return
			}
			continue
		}
		for {
			ev, err := stream.Recv()
			if err != nil {
				if ctx.Err() == nil {
					logger.Warn("push stream ended", zap.Error(err))
				}
				break
			}
			// A chat-access-changed event is an infra signal, not an in-app event:
			// resolve the recipient's Telegram identity and current eligibility and
			// push the chat-gate command to the bot, without fanning it out to clients.
			if ev.GetKind() == chatAccessChangedKind {
				if bot != nil {
					go deliverChatGate(ctx, backend, bot, ev.GetUserId(), logger)
				}
				continue
			}
			hub.Publish(push.Event{
				UserID:  ev.GetUserId(),
				Kind:    ev.GetKind(),
				Payload: ev.GetPayload(),
				EventID: ev.GetEventId(),
			})
			// Out-of-app fallback: when the recipient has no live in-app stream,
			// deliver the event over the bot-link. Done in a goroutine so a slow
			// target lookup never stalls the in-app firehose.
			if bot != nil && connector.OutOfAppKind(ev.GetKind()) && !hub.HasSubscribers(ev.GetUserId()) {
				go deliverOutOfApp(ctx, backend, bot, ev.GetUserId(), ev.GetKind(), ev.GetPayload(), logger)
			}
		}
		if !sleep(ctx, pushReconnectDelay) {
			return
		}
	}
}

// deliverOutOfApp resolves the recipient's push target and, when they have a
// Telegram identity and have not confined notifications to the app, asks the
// connector to deliver the event. It is best-effort: every failure is logged and
// dropped (the in-app stream remains the primary channel).
func deliverOutOfApp(ctx context.Context, backend *backendclient.Client, bot *botlink.Hub, userID, kind string, payload []byte, logger *zap.Logger) {
	target, err := backend.PushTarget(ctx, userID)
	if err != nil {
		logger.Warn("push target lookup failed", zap.String("user_id", userID), zap.Error(err))
		return
	}
	if !connector.DeliverToTarget(target.ExternalID, target.NotificationsInAppOnly) {
		return
	}
	// Fire-and-forget down the bot-link; the bot renders the message in the
	// recipient's interface language and is dropped (best-effort) if no bot is up.
	bot.Send(botlink.NotifyCommand(target.ExternalID, kind, payload, target.Language))
}

// chatAccessChangedKind is the backend event signalling that a player's moderated-chat
// write eligibility may have changed; the gateway turns it into a bot-link chat-gate
// command rather than an in-app event (it mirrors notify.KindChatAccessChanged).
const chatAccessChangedKind = "chat_access_changed"

// deliverChatGate resolves a chat-access-changed event to the recipient's Telegram
// identity and current eligibility and pushes the chat-gate command to the bot. It is
// best-effort: a recipient with no Telegram identity is skipped, and a resolve failure
// is logged and dropped (the next moderation action, or a re-join, re-applies the gate).
func deliverChatGate(ctx context.Context, backend *backendclient.Client, bot *botlink.Hub, userID string, logger *zap.Logger) {
	res, err := backend.ChatAccessByUser(ctx, userID)
	if err != nil {
		logger.Warn("chat-gate resolve failed", zap.String("user_id", userID), zap.Error(err))
		return
	}
	if res.ExternalID == "" {
		return // no Telegram identity, nothing to gate
	}
	bot.Send(botlink.ChatGateCommand(res.ExternalID, res.Eligible))
}

// sleep waits for d or until ctx is cancelled, reporting whether it waited the
// full duration.
func sleep(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// 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()
}