8878711cf3
- GATEWAY_MAX_BODY_BYTES (1 MiB): connect WithReadMaxBytes + http.MaxBytesReader
on the public mux; explicit http2.Server MaxConcurrentStreams/IdleTimeout and
an http.Server ReadHeaderTimeout (R2 report follow-up).
- gateway_rate_limited_total{class} counter, Debug per rejection, a rejection
tracker drained every 30 s into a Warn summary per key and a report POST to
/api/v1/internal/ratelimit/report (feeds the admin view + auto-flag).
- The dead AdminPerMinute/AdminBurst policy now guards the /_gm mount (429),
ahead of its Basic-Auth.
- resolve() logs the cause of infra session-resolve failures at Warn (the
transient unauthenticated dips from the R2 run); unknown tokens stay silent.
140 lines
4.7 KiB
Go
140 lines
4.7 KiB
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"
|
|
)
|
|
|
|
// 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)
|
|
}
|
|
return &Client{
|
|
baseURL: strings.TrimRight(httpURL, "/"),
|
|
http: &http.Client{Timeout: timeout},
|
|
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 (R3).
|
|
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)
|
|
}
|