// Package backendclient is the gateway's typed client for the backend: REST/JSON // for synchronous operations (injecting X-User-ID) and a gRPC subscription for // the live push stream. The response structs mirror the backend's JSON DTOs; the // transcode layer turns them into FlatBuffers for the client. package backendclient import ( "bytes" "context" "encoding/json" "fmt" "io" "net/http" "strings" "time" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "scrabble/gateway/internal/ratelimit" pushv1 "scrabble/pkg/proto/push/v1" ) // backendMaxIdleConns sizes the REST keep-alive pool to the single backend host. The // default transport caps idle connections per host at 2 (http.DefaultMaxIdleConnsPerHost), // which — since every synchronous client call proxies to that one host — forces a fresh // TCP connection (and a lingering TIME_WAIT socket) for almost every request under load. // That connection churn burns gateway CPU and exhausts ephemeral ports at scale, all // while the backend itself sits near-idle. Pooling the connections lets them be reused. // // The stress harness measured the effect at 500 concurrent players: the churn collapsed // from ~26 500 TIME_WAIT sockets to ~0 and peak gateway CPU from ~1.75 to ~0.26 cores, // with the pool settling at ~225 live connections. 512 keeps ~2x headroom over that // observed peak so a burst never re-caps the pool. See loadtest/REPORT.md. const backendMaxIdleConns = 512 // Client calls the backend's REST API and opens its push gRPC stream. type Client struct { baseURL string http *http.Client conn *grpc.ClientConn push pushv1.PushClient } // New dials the backend push gRPC endpoint and prepares the REST client. The // backend lives on a trusted network segment, so the gRPC connection uses // insecure (plaintext) transport credentials (ARCHITECTURE.md §12). func New(httpURL, grpcAddr string, timeout time.Duration) (*Client, error) { conn, err := grpc.NewClient(grpcAddr, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithStatsHandler(otelgrpc.NewClientHandler()), ) if err != nil { return nil, fmt.Errorf("backendclient: dial push %s: %w", grpcAddr, err) } // Clone the default transport (keeping its proxy, dialer and timeouts) and widen the // idle pool so REST calls to the backend reuse connections instead of churning them. transport := http.DefaultTransport.(*http.Transport).Clone() transport.MaxIdleConns = backendMaxIdleConns transport.MaxIdleConnsPerHost = backendMaxIdleConns return &Client{ baseURL: strings.TrimRight(httpURL, "/"), http: &http.Client{Timeout: timeout, Transport: transport}, conn: conn, push: pushv1.NewPushClient(conn), }, nil } // Close releases the gRPC connection. func (c *Client) Close() error { return c.conn.Close() } // APIError carries a backend error response so the transcode layer can surface a // stable result code to the client. type APIError struct { Status int Code string Message string } func (e *APIError) Error() string { return fmt.Sprintf("backend %d (%s): %s", e.Status, e.Code, e.Message) } // do performs one REST call. userID, when non-empty, is forwarded as X-User-ID; // clientIP, when non-empty, as X-Forwarded-For (for chat moderation). A non-2xx // response is returned as an *APIError carrying the backend error code. func (c *Client) do(ctx context.Context, method, path, userID, clientIP string, body, out any) error { var reader io.Reader if body != nil { raw, err := json.Marshal(body) if err != nil { return fmt.Errorf("backendclient: marshal request: %w", err) } reader = bytes.NewReader(raw) } req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, reader) if err != nil { return fmt.Errorf("backendclient: new request: %w", err) } req.Header.Set("Content-Type", "application/json") if userID != "" { req.Header.Set("X-User-ID", userID) } if clientIP != "" { req.Header.Set("X-Forwarded-For", clientIP) } resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("backendclient: %s %s: %w", method, path, err) } defer func() { _ = resp.Body.Close() }() data, err := io.ReadAll(resp.Body) if err != nil { return fmt.Errorf("backendclient: read response: %w", err) } if resp.StatusCode >= http.StatusMultipleChoices { return parseAPIError(resp.StatusCode, data) } if out != nil { if err := json.Unmarshal(data, out); err != nil { return fmt.Errorf("backendclient: decode response: %w", err) } } return nil } // getRaw performs one REST GET, returning the raw (un-decoded) response body and // the value of respHeader (e.g. Cache-Control). userID, when non-empty, is // forwarded as X-User-ID. A non-2xx response is returned as an *APIError. Unlike // do it does not JSON-decode, so it carries a binary payload such as a dictionary // blob. func (c *Client) getRaw(ctx context.Context, path, userID, respHeader string) ([]byte, string, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) if err != nil { return nil, "", fmt.Errorf("backendclient: new request: %w", err) } if userID != "" { req.Header.Set("X-User-ID", userID) } resp, err := c.http.Do(req) if err != nil { return nil, "", fmt.Errorf("backendclient: GET %s: %w", path, err) } defer func() { _ = resp.Body.Close() }() data, err := io.ReadAll(resp.Body) if err != nil { return nil, "", fmt.Errorf("backendclient: read response: %w", err) } if resp.StatusCode >= http.StatusMultipleChoices { return nil, "", parseAPIError(resp.StatusCode, data) } return data, resp.Header.Get(respHeader), nil } // parseAPIError extracts the backend's {error:{code,message}} envelope. func parseAPIError(status int, data []byte) *APIError { var env struct { Error struct { Code string `json:"code"` Message string `json:"message"` } `json:"error"` } if err := json.Unmarshal(data, &env); err == nil && env.Error.Code != "" { return &APIError{Status: status, Code: env.Error.Code, Message: env.Error.Message} } return &APIError{Status: status, Code: "backend_error", Message: strings.TrimSpace(string(data))} } // SubscribePush opens the backend live-event stream. func (c *Client) SubscribePush(ctx context.Context, gatewayID string) (grpc.ServerStreamingClient[pushv1.Event], error) { return c.push.Subscribe(ctx, &pushv1.SubscribeRequest{GatewayId: gatewayID}) } // ReportRateLimited posts the gateway's periodic rate-limiter rejection summary // to the backend, which feeds the admin console's throttled view and the // high-rate auto-flag. The endpoint carries no user identity: like // sessions/resolve it rides the trusted internal segment. func (c *Client) ReportRateLimited(ctx context.Context, windowSeconds int, entries []ratelimit.Rejection) error { body := struct { WindowSeconds int `json:"window_seconds"` Entries []ratelimit.Rejection `json:"entries"` }{WindowSeconds: windowSeconds, Entries: entries} return c.do(ctx, http.MethodPost, "/api/v1/internal/ratelimit/report", "", "", body, nil) } // SyncBans reports the gateway's currently-active IP bans to the backend and // returns the IPs an operator has marked for unban since the previous sync. It is // the ban mirror of ReportRateLimited plus the manual-unban backchannel: the // backend renders the active set in the admin console and drains the operator's // unban requests into the response. Like the rejection report it carries no user // identity and rides the trusted internal segment. func (c *Client) SyncBans(ctx context.Context, active []ratelimit.Ban) ([]string, error) { body := struct { Active []ratelimit.Ban `json:"active"` }{Active: active} var out struct { Unban []string `json:"unban"` } if err := c.do(ctx, http.MethodPost, "/api/v1/internal/bans/sync", "", "", body, &out); err != nil { return nil, err } return out.Unban, nil }