scrabble-game/gateway/internal/backendclient/client.go

// 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
	// dl serves the export downloads: the backend may wait on the render sidecar
	// for several seconds, so these calls get their own, longer deadline than the
	// ordinary REST budget.
	dl   *http.Client
	conn *grpc.ClientConn
	push pushv1.PushClient
}

// exportDownloadTimeout bounds one export-download fetch end to end (the backend's
// own sidecar budget is 15s).
const exportDownloadTimeout = 30 * time.Second

// 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},
		dl:      &http.Client{Timeout: exportDownloadTimeout, Transport: transport},
		conn:    conn,
		push:    pushv1.NewPushClient(conn),
	}, nil
}

// ExportDownload fetches a signed finished-game export artifact from the backend's
// public group, forwarding the caller's public Host for the image footer. It returns
// the bytes with the backend's Content-Type and Content-Disposition. rest is the
// public path suffix after /dl (e.g. "/<game>/<kind>?e=…&s=…").
func (c *Client) ExportDownload(ctx context.Context, rest, publicHost string) ([]byte, string, string, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/api/v1/public/dl"+rest, nil)
	if err != nil {
		return nil, "", "", fmt.Errorf("backendclient: new request: %w", err)
	}
	if publicHost != "" {
		req.Header.Set("X-Public-Host", publicHost)
	}
	resp, err := c.dl.Do(req)
	if err != nil {
		return nil, "", "", fmt.Errorf("backendclient: GET export: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()
	data, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, "", "", fmt.Errorf("backendclient: read export: %w", err)
	}
	if resp.StatusCode >= http.StatusMultipleChoices {
		return nil, "", "", parseAPIError(resp.StatusCode, data)
	}
	return data, resp.Header.Get("Content-Type"), resp.Header.Get("Content-Disposition"), 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)
}

// platformCtxKey types the request-context slot the trusted X-Platform value rides in.
type platformCtxKey struct{}

// WithPlatform returns a copy of ctx carrying the trusted platform header value
// ("<kind>/<subtype>", e.g. "vk/ios") that do and getRaw inject as X-Platform on the
// outbound backend request. An empty platform (an untrusted session) leaves ctx
// unchanged, so no header is sent and the backend treats the request as untrusted.
func WithPlatform(ctx context.Context, platform string) context.Context {
	if platform == "" {
		return ctx
	}
	return context.WithValue(ctx, platformCtxKey{}, platform)
}

// platformFromContext returns the platform header value stored by WithPlatform, or
// an empty string when none is present.
func platformFromContext(ctx context.Context) string {
	p, _ := ctx.Value(platformCtxKey{}).(string)
	return p
}

// clientIPCtxKey types the request-context slot the originating client IP rides in.
type clientIPCtxKey struct{}

// WithClientIP returns a copy of ctx carrying the originating client IP that do injects as
// X-Forwarded-For on every downstream backend request — so the backend records the real caller
// (the account's last-login IP, chat/feedback moderation, the access log) rather than the gateway's
// own connection. An empty IP leaves ctx unchanged, so no header is sent.
func WithClientIP(ctx context.Context, ip string) context.Context {
	if ip == "" {
		return ctx
	}
	return context.WithValue(ctx, clientIPCtxKey{}, ip)
}

// clientIPFromContext returns the client IP stored by WithClientIP, or an empty string when none.
func clientIPFromContext(ctx context.Context) string {
	ip, _ := ctx.Value(clientIPCtxKey{}).(string)
	return ip
}

// 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); the trusted
// platform carried on ctx (see WithPlatform), when present, as X-Platform. 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)
	}
	// The client IP rides X-Forwarded-For so the backend records the real caller. It comes from the
	// explicit param (chat/feedback) or, for every other call, the request context (WithClientIP, set
	// once per request in the Connect edge) — so the backend never falls back to the gateway's own
	// connection address.
	if clientIP == "" {
		clientIP = clientIPFromContext(ctx)
	}
	if clientIP != "" {
		req.Header.Set("X-Forwarded-For", clientIP)
	}
	if p := platformFromContext(ctx); p != "" {
		req.Header.Set("X-Platform", p)
	}
	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)
	}
	if p := platformFromContext(ctx); p != "" {
		req.Header.Set("X-Platform", p)
	}
	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
}