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
	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
}

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