mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
ce17a45afb
Closes the 'live duplicate/conflict alerting also remains' gap noted
in §8.10: a durable domain.ErrConflict/ErrResultConflict rejection on
/v1/servers/{id}/{register,connect,disconnect,shutdown,result} was
already logged as a structured 'conflict' stage event, but had no
Prometheus signal distinct from the generic 4xx-class counter, which
also catches ordinary client noise (malformed bodies, expired
tokens). A real duplicate registration, raced reconnect, or replayed
result would have been invisible to alerting until someone went
looking through logs.
observability.Metrics gains ObserveServerConflict(kind), a bounded
counter keyed to serverMutation's own five routes (an unrecognized
kind folds into "other", so a caller mistake can't grow the label
set), exported as cosmic_clash_api_server_conflicts_total. Wired at
each of serverMutation's four conflict branches in server/api/service.go.
deploy/observability/prometheus-rules.yaml adds
CosmicClashControlPlaneServerConflicts, mirroring the existing
allocator quota-denial alert shape, firing on >3 conflicts of one
kind in 15 minutes.
Verified: go build/vet/test -race clean across every server package;
new unit tests cover per-kind counting, the bounded 'other' fallback,
the counter's absence until first observed, and a nil-receiver no-op;
a service-level test proves a real register conflict is exported
through the live /metrics endpoint. scripts/verify_observability_manifests.py
passes against the edited rules file.
Remaining, and explicitly out of scope here: this alert has only been
validated statically, never against a live Prometheus/Alertmanager
firing on real traffic — that requires the same live cluster this
sandbox has never had.
1256 lines
47 KiB
Go
1256 lines
47 KiB
Go
// Package api exposes the small authenticated HTTP boundary around domain
|
|
// policies. Persistent adapters can replace the in-memory dependencies without
|
|
// changing request authentication or validation rules.
|
|
package api
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
|
"github.com/cosmic-clash/cosmic-clash/server/observability"
|
|
)
|
|
|
|
const maxBodyBytes = 8 << 10
|
|
|
|
type CandidateProvider func(playerID, ticketID string) (domain.Candidate, error)
|
|
type CandidateProviderV2 func(playerID, ticketID string, spec domain.QueueSpec) (domain.Candidate, error)
|
|
type ProbeProvider func(playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error)
|
|
type ProbeRecorder interface {
|
|
RecordProbe(context.Context, string, string, time.Duration, time.Time) error
|
|
}
|
|
type WorkloadVerifier func(string, time.Time) (domain.WorkloadBinding, error)
|
|
type RankedProfileProvider interface {
|
|
Get(context.Context, string) (domain.RankedProfile, bool, error)
|
|
}
|
|
type ResultSubmitter interface {
|
|
SubmitResult(context.Context, string, domain.MatchResult, domain.WorkloadBinding, []byte, time.Time) error
|
|
}
|
|
type ServerRegistrar interface {
|
|
RegisterServer(context.Context, domain.WorkloadBinding, int, bool, string, time.Time) error
|
|
}
|
|
type ServerShutdowner interface {
|
|
ShutdownServer(context.Context, domain.WorkloadBinding, string, string, time.Time) error
|
|
}
|
|
type ServerConnectionRecorder interface {
|
|
ClaimPlayerConnection(context.Context, domain.WorkloadBinding, string, uint64, string, time.Time) (uint64, error)
|
|
RecordPlayerDisconnected(context.Context, domain.WorkloadBinding, string, uint64, string, time.Time) error
|
|
}
|
|
|
|
type QueueBackend interface {
|
|
Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error)
|
|
Heartbeat(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error)
|
|
Cancel(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error)
|
|
Get(context.Context, string, string, time.Time) (domain.QueueTicket, error)
|
|
}
|
|
|
|
// CandidateIndex is a transient projection of durable queue ownership. Index
|
|
// failures must never change the result of an already successful mutation.
|
|
type CandidateIndex interface {
|
|
Upsert(context.Context, domain.Candidate) error
|
|
Remove(context.Context, string) error
|
|
}
|
|
|
|
type SessionBackend interface {
|
|
Authenticate(context.Context, string, string, time.Time) (domain.Session, error)
|
|
}
|
|
type SteamLoginProvider interface {
|
|
Authenticate(context.Context, string, time.Time) (domain.VerifiedIdentity, error)
|
|
}
|
|
type SessionIssuer interface {
|
|
Issue(context.Context, string, time.Duration, time.Time) (domain.Session, string, error)
|
|
}
|
|
type ProposalBackend interface {
|
|
Get(context.Context, string, string, time.Time) (domain.Proposal, error)
|
|
}
|
|
type ProposalMutationBackend interface {
|
|
Respond(context.Context, string, string, string, bool, uint64, time.Time) (domain.Proposal, error)
|
|
}
|
|
type ProposalPromoter interface {
|
|
Promote(context.Context, domain.Proposal, time.Time) error
|
|
}
|
|
type ProposalPromoterFunc func(context.Context, domain.Proposal, time.Time) error
|
|
|
|
func (f ProposalPromoterFunc) Promote(ctx context.Context, proposal domain.Proposal, now time.Time) error {
|
|
return f(ctx, proposal, now)
|
|
}
|
|
|
|
type AssignmentView struct {
|
|
MatchID string `json:"match_id"`
|
|
ServerID string `json:"server_id"`
|
|
PlayerID string `json:"player_id"`
|
|
Slot int `json:"slot"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
ProtocolVersion int `json:"protocol_version"`
|
|
Transport string `json:"transport"`
|
|
JoinAuthorisation string `json:"join_authorisation"`
|
|
Endpoint string `json:"endpoint"`
|
|
// Revision is routing metadata for the event stream, not part of the v1
|
|
// assignment response. Keeping it alongside the durable view prevents the
|
|
// REST recovery boundary from emitting a synthetic revision zero.
|
|
Revision uint64 `json:"-"`
|
|
}
|
|
|
|
type AssignmentProvider func(context.Context, string, string, time.Time) (AssignmentView, error)
|
|
type RosterProvider func(context.Context, domain.WorkloadBinding, time.Time) ([][]byte, error)
|
|
type ReadinessCheck func(context.Context) error
|
|
|
|
type Service struct {
|
|
Sessions *domain.SessionStore
|
|
SessionBackend SessionBackend
|
|
SessionIssuer SessionIssuer
|
|
SteamLogin SteamLoginProvider
|
|
Queue *domain.Queue
|
|
Candidate CandidateProvider
|
|
CandidateV2 CandidateProviderV2
|
|
QueueBackend QueueBackend
|
|
CandidateIndex CandidateIndex
|
|
Probe ProbeProvider
|
|
ProbeRecorder ProbeRecorder
|
|
WorkloadVerify WorkloadVerifier
|
|
ResultSubmitter ResultSubmitter
|
|
ServerRegistrar ServerRegistrar
|
|
ServerShutdowner ServerShutdowner
|
|
ServerConnections ServerConnectionRecorder
|
|
Assignment AssignmentProvider
|
|
Roster RosterProvider
|
|
Now func() time.Time
|
|
Proposals map[string]*domain.Proposal
|
|
ProposalBackend ProposalBackend
|
|
ProposalPromoter ProposalPromoter
|
|
RankedProfiles map[string]domain.RankedProfile
|
|
RankedProfileProvider RankedProfileProvider
|
|
TierPolicy domain.TierPolicy
|
|
RateLimiter *RateLimiter
|
|
ClientIPs *ClientIPResolver
|
|
Admission AdmissionController
|
|
ReadinessCheck ReadinessCheck
|
|
// Log receives a credential-safe structured event for lifecycle-relevant
|
|
// reads and mutations. Nil
|
|
// is a valid, silent no-op -- every call site must stay optional so
|
|
// existing Service literals that don't set it keep working unchanged.
|
|
Log func(observability.Event)
|
|
Metrics *observability.Metrics
|
|
proposalMu sync.Mutex
|
|
eventsMu sync.Mutex
|
|
events *eventHub
|
|
}
|
|
|
|
// rankedProfileFor prefers the durable RankedProfileProvider when set,
|
|
// falling back to the in-memory RankedProfiles map for existing tests/direct
|
|
// Service literals that construct it that way. Both return the same
|
|
// (profile, exists) shape either way, so callers don't need to know which
|
|
// source answered.
|
|
func (s *Service) rankedProfileFor(ctx context.Context, playerID string) (domain.RankedProfile, bool, error) {
|
|
if s.RankedProfileProvider != nil {
|
|
return s.RankedProfileProvider.Get(ctx, playerID)
|
|
}
|
|
profile, exists := s.RankedProfiles[playerID]
|
|
return profile, exists, nil
|
|
}
|
|
|
|
// logEvent is a nil-safe wrapper so call sites never need their own guard.
|
|
func (s *Service) logEvent(event observability.Event) {
|
|
if s.Log != nil {
|
|
s.Log(event)
|
|
}
|
|
}
|
|
|
|
// logQueueOutcome logs a queue-ticket mutation's result: the ticket's
|
|
// resulting state on success, or "rejected" on a domain error. It never logs
|
|
// the error text itself -- domain errors here are not documented as
|
|
// credential-free, and the stage name already tells an operator what to look
|
|
// up (the ticket ID, still recorded either way).
|
|
func (s *Service) logQueueOutcome(event, ticketID string, ticket domain.QueueTicket, err error, now time.Time) {
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: event, QueueID: ticketID, Stage: "rejected", OccurredAt: now})
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: event, QueueID: ticket.TicketID, Stage: strings.ToLower(string(ticket.State)), OccurredAt: now})
|
|
}
|
|
|
|
// logProposalOutcome mirrors logQueueOutcome for proposal accept/decline.
|
|
func (s *Service) logProposalOutcome(proposalID string, proposal domain.Proposal, err error, now time.Time) {
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: "proposal_response", ProposalID: proposalID, Stage: "rejected", OccurredAt: now})
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "proposal_response", ProposalID: proposal.ProposalID, Stage: strings.ToLower(string(proposal.State)), OccurredAt: now})
|
|
}
|
|
|
|
func (s *Service) Handler() http.Handler {
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("/healthz", s.health)
|
|
mux.HandleFunc("/readyz", s.ready)
|
|
mux.HandleFunc("/metrics", s.metrics)
|
|
mux.HandleFunc("/v1/session/steam", s.steamSession)
|
|
mux.HandleFunc("/v1/queue", s.queueCreate)
|
|
mux.HandleFunc("/v1/queue/", s.queueMutation)
|
|
mux.HandleFunc("/v1/proposals/", s.proposalMutation)
|
|
mux.HandleFunc("/v1/assignments/", s.assignment)
|
|
mux.HandleFunc("/v1/profile/ranked", s.rankedProfile)
|
|
mux.HandleFunc("/v1/probes/", s.probe)
|
|
mux.HandleFunc("/v1/events", s.controlPlaneEvent)
|
|
mux.HandleFunc("/v1/servers/", s.serverMutation)
|
|
// The public contract is served below /api/v1. Keep the original /v1
|
|
// routes for the Godot client while exposing the documented names.
|
|
mux.HandleFunc("/api/v1/session/steam", s.steamSession)
|
|
mux.HandleFunc("/api/v1/profile", s.profile)
|
|
mux.HandleFunc("/api/v1/queue/tickets", s.contractQueueCreate)
|
|
mux.HandleFunc("/api/v1/queue/tickets/", s.contractQueueMutation)
|
|
mux.HandleFunc("/api/v1/proposals/", s.contractProposalMutation)
|
|
mux.HandleFunc("/api/v1/assignments/", s.contractAssignment)
|
|
mux.HandleFunc("/api/v1/events", s.controlPlaneEvent)
|
|
mux.HandleFunc("/api/v1/servers/", s.contractServerMutation)
|
|
var handler http.Handler = mux
|
|
if s.RateLimiter != nil {
|
|
handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path == "/healthz" || r.URL.Path == "/readyz" || r.URL.Path == "/metrics" {
|
|
mux.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
if !s.RateLimiter.AllowKeys(requestRateKeys(r, s.ClientIPs), s.now()) {
|
|
writeError(w, http.StatusTooManyRequests, "rate_limited")
|
|
return
|
|
}
|
|
mux.ServeHTTP(w, r)
|
|
})
|
|
}
|
|
if s.Admission != nil {
|
|
admissionHandler := handler
|
|
handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if operation := admissionOperation(r.URL.Path, r.Method); operation != "" && !s.Admission.Allow(operation) {
|
|
writeError(w, http.StatusServiceUnavailable, "service_degraded")
|
|
return
|
|
}
|
|
admissionHandler.ServeHTTP(w, r)
|
|
})
|
|
}
|
|
if s.Metrics == nil {
|
|
return handler
|
|
}
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path == "/metrics" || r.URL.Path == "/healthz" || r.URL.Path == "/readyz" || strings.HasSuffix(r.URL.Path, "/events") {
|
|
handler.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
started := time.Now()
|
|
recorder := &statusRecorder{ResponseWriter: w}
|
|
handler.ServeHTTP(recorder, r)
|
|
code := recorder.code
|
|
if code == 0 {
|
|
code = http.StatusOK
|
|
}
|
|
s.Metrics.ObserveAPI(metricOperation(r.URL.Path), code, time.Since(started))
|
|
})
|
|
}
|
|
|
|
type statusRecorder struct {
|
|
http.ResponseWriter
|
|
code int
|
|
}
|
|
|
|
func (w *statusRecorder) WriteHeader(code int) {
|
|
w.code = code
|
|
w.ResponseWriter.WriteHeader(code)
|
|
}
|
|
|
|
func (w *statusRecorder) Write(body []byte) (int, error) {
|
|
if w.code == 0 {
|
|
w.WriteHeader(http.StatusOK)
|
|
}
|
|
return w.ResponseWriter.Write(body)
|
|
}
|
|
|
|
func (s *Service) metrics(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet || s.Metrics == nil {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
|
|
_ = s.Metrics.WritePrometheus(w)
|
|
}
|
|
|
|
func metricOperation(path string) string {
|
|
switch {
|
|
case strings.Contains(path, "/queue"):
|
|
return "queue"
|
|
case strings.Contains(path, "/proposals"):
|
|
return "proposal"
|
|
case strings.Contains(path, "/assignments"):
|
|
return "assignment"
|
|
case strings.Contains(path, "ranked"):
|
|
return "ranked_profile"
|
|
case strings.Contains(path, "/profile"):
|
|
return "profile"
|
|
case strings.Contains(path, "/servers"):
|
|
return "server"
|
|
case strings.Contains(path, "/session"):
|
|
return "session"
|
|
case strings.Contains(path, "/probes"):
|
|
return "probe"
|
|
default:
|
|
return "other"
|
|
}
|
|
}
|
|
|
|
type steamSessionRequest struct {
|
|
WebAPITicket string `json:"web_api_ticket"`
|
|
}
|
|
|
|
func (s *Service) steamSession(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
if s.SteamLogin == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "auth_unavailable")
|
|
return
|
|
}
|
|
var input steamSessionRequest
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if input.WebAPITicket == "" || len(input.WebAPITicket) > 4096 {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return
|
|
}
|
|
now := s.now()
|
|
identity, err := s.SteamLogin.Authenticate(r.Context(), input.WebAPITicket, now)
|
|
if err != nil || identity.PlayerID == "" || identity.SteamID == "" {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return
|
|
}
|
|
var session domain.Session
|
|
var token string
|
|
if s.SessionIssuer != nil {
|
|
session, token, err = s.SessionIssuer.Issue(r.Context(), identity.PlayerID, time.Hour, now)
|
|
} else if s.Sessions != nil {
|
|
session, token, err = s.Sessions.Issue(identity.PlayerID, time.Hour, now)
|
|
} else {
|
|
writeError(w, http.StatusServiceUnavailable, "auth_unavailable")
|
|
return
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusServiceUnavailable, "auth_unavailable")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"player_id": session.PlayerID, "expires_at": session.ExpiresAt, "access_token": session.SessionID + ":" + token})
|
|
}
|
|
|
|
func (s *Service) health(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
|
}
|
|
|
|
func (s *Service) ready(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
if s.ReadinessCheck == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "not_ready")
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(r.Context(), time.Second)
|
|
defer cancel()
|
|
if err := s.ReadinessCheck(ctx); err != nil {
|
|
writeError(w, http.StatusServiceUnavailable, "not_ready")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
|
}
|
|
|
|
type queueCreateRequest struct {
|
|
TicketID string `json:"ticket_id"`
|
|
Playlist string `json:"playlist"`
|
|
ClientBuild string `json:"client_build"`
|
|
ProtocolVersion int `json:"protocol_version"`
|
|
}
|
|
type queueResponse struct {
|
|
TicketID string `json:"ticket_id"`
|
|
PlayerID string `json:"player_id"`
|
|
ProposalID string `json:"proposal_id,omitempty"`
|
|
MatchID string `json:"match_id,omitempty"`
|
|
State string `json:"state"`
|
|
Revision uint64 `json:"revision"`
|
|
EnqueuedAt time.Time `json:"enqueued_at"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
Playlist string `json:"playlist"`
|
|
}
|
|
|
|
func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
if (s.Queue == nil && s.QueueBackend == nil) || (s.QueueBackend == nil && s.Candidate == nil && s.CandidateV2 == nil) {
|
|
writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
|
|
return
|
|
}
|
|
var input queueCreateRequest
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if input.TicketID == "" || (input.Playlist != string(domain.Casual) && input.Playlist != string(domain.Ranked)) || input.ClientBuild == "" || len(input.ClientBuild) > 128 || input.ProtocolVersion < 1 {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return
|
|
}
|
|
key := r.Header.Get("Idempotency-Key")
|
|
if len(key) < 16 || len(key) > 128 {
|
|
writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
|
|
return
|
|
}
|
|
now := s.now()
|
|
spec := domain.QueueSpec{Playlist: domain.Playlist(input.Playlist), ClientBuild: input.ClientBuild, ProtocolVersion: input.ProtocolVersion}
|
|
if s.QueueBackend != nil {
|
|
ticket, err := s.QueueBackend.Create(r.Context(), playerID, input.TicketID, key, spec, now)
|
|
if err != nil {
|
|
s.logQueueOutcome("queue_create", input.TicketID, ticket, err, now)
|
|
writeDomainError(w, err)
|
|
return
|
|
}
|
|
s.logQueueOutcome("queue_create", input.TicketID, ticket, nil, now)
|
|
s.projectCandidate(r.Context(), ticket)
|
|
s.publishTicketEvent(ticket, now)
|
|
writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
|
|
return
|
|
}
|
|
var candidate domain.Candidate
|
|
var err error
|
|
if s.CandidateV2 != nil {
|
|
candidate, err = s.CandidateV2(playerID, input.TicketID, spec)
|
|
} else {
|
|
candidate, err = s.Candidate(playerID, input.TicketID)
|
|
// Legacy providers predate queue compatibility metadata. The API has
|
|
// validated the request; keep the resulting projection self-describing.
|
|
candidate.Playlist = spec.Playlist
|
|
candidate.ClientBuild = spec.ClientBuild
|
|
candidate.ProtocolVersion = spec.ProtocolVersion
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusUnprocessableEntity, "candidate_unavailable")
|
|
return
|
|
}
|
|
if candidate.PlayerID != playerID || candidate.TicketID != input.TicketID || candidate.Playlist != spec.Playlist || candidate.ClientBuild != spec.ClientBuild || candidate.ProtocolVersion != spec.ProtocolVersion {
|
|
writeError(w, http.StatusUnprocessableEntity, "candidate_mismatch")
|
|
return
|
|
}
|
|
ticket, err := s.Queue.Create(playerID, input.TicketID, key, candidate, now)
|
|
if err != nil {
|
|
s.logQueueOutcome("queue_create", input.TicketID, ticket, err, now)
|
|
writeDomainError(w, err)
|
|
return
|
|
}
|
|
s.logQueueOutcome("queue_create", input.TicketID, ticket, nil, now)
|
|
s.projectCandidate(r.Context(), ticket)
|
|
s.publishTicketEvent(ticket, now)
|
|
writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
|
|
}
|
|
|
|
func (s *Service) projectCandidate(ctx context.Context, ticket domain.QueueTicket) {
|
|
if s.CandidateIndex != nil {
|
|
_ = s.CandidateIndex.Upsert(ctx, ticket.Candidate)
|
|
}
|
|
}
|
|
|
|
func (s *Service) removeCandidate(ctx context.Context, ticketID string) {
|
|
if s.CandidateIndex != nil {
|
|
_ = s.CandidateIndex.Remove(ctx, ticketID)
|
|
}
|
|
}
|
|
|
|
func (s *Service) contractQueueCreate(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
s.queueCreate(w, r)
|
|
return
|
|
}
|
|
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxBodyBytes))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return
|
|
}
|
|
var fields map[string]json.RawMessage
|
|
if json.Unmarshal(body, &fields) == nil {
|
|
if rawTicketID, exists := fields["ticket_id"]; exists {
|
|
var ticketID string
|
|
if json.Unmarshal(rawTicketID, &ticketID) != nil || !controlPlaneResourceIDRE.MatchString(ticketID) {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return
|
|
}
|
|
}
|
|
// Ticket IDs are server-assigned for the public contract. Deriving one
|
|
// from the authenticated request's idempotency material makes retries
|
|
// converge on the same domain command without persisting adapter state.
|
|
digest := sha256.Sum256([]byte(r.Header.Get("Authorization") + "\x00" + r.Header.Get("Idempotency-Key")))
|
|
id := hex.EncodeToString(digest[:])
|
|
if _, exists := fields["ticket_id"]; !exists {
|
|
fields["ticket_id"] = json.RawMessage(strconv.Quote(id))
|
|
body, _ = json.Marshal(fields)
|
|
}
|
|
}
|
|
r.Body = io.NopCloser(bytes.NewReader(body))
|
|
s.queueCreate(w, r)
|
|
}
|
|
|
|
func (s *Service) contractQueueMutation(w http.ResponseWriter, r *http.Request) {
|
|
path := strings.TrimPrefix(r.URL.Path, "/api/v1/queue/tickets/")
|
|
parts := strings.Split(path, "/")
|
|
if path == "" || len(parts) > 2 || !controlPlaneResourceIDRE.MatchString(parts[0]) || (len(parts) == 2 && parts[1] != "heartbeat") {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
clone := r.Clone(r.Context())
|
|
clone.URL.Path = "/v1/queue/" + parts[0]
|
|
if len(parts) == 2 {
|
|
clone.URL.Path += "/heartbeat"
|
|
}
|
|
if r.Method == http.MethodDelete {
|
|
if len(parts) != 1 {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
clone.Method = http.MethodPost
|
|
clone.URL.Path += "/cancel"
|
|
clone.Header.Set("X-Contract-Delete", "1")
|
|
}
|
|
s.queueMutation(w, clone)
|
|
}
|
|
|
|
func (s *Service) contractProposalMutation(w http.ResponseWriter, r *http.Request) {
|
|
path := strings.TrimPrefix(r.URL.Path, "/api/v1/proposals/")
|
|
parts := strings.Split(path, "/")
|
|
if path == "" || len(parts) > 2 || !controlPlaneResourceIDRE.MatchString(parts[0]) {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
clone := r.Clone(r.Context())
|
|
clone.URL.Path = "/v1/proposals/" + path
|
|
s.proposalMutation(w, clone)
|
|
}
|
|
|
|
func (s *Service) contractAssignment(w http.ResponseWriter, r *http.Request) {
|
|
path := strings.TrimPrefix(r.URL.Path, "/api/v1/assignments/")
|
|
if path == "" || strings.Contains(path, "/") || !controlPlaneResourceIDRE.MatchString(path) {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
clone := r.Clone(r.Context())
|
|
clone.URL.Path = "/v1/assignments/" + path
|
|
s.assignment(w, clone)
|
|
}
|
|
|
|
func (s *Service) contractServerMutation(w http.ResponseWriter, r *http.Request) {
|
|
// Unlike contractAssignment, the documented shape here is two segments
|
|
// (/servers/{serverId}/{result|register|roster|connect|disconnect|shutdown}) — rejecting
|
|
// any "/" would 404 every real call. Delegate shape validation to
|
|
// serverMutation, which already enforces the exact operation allowlist.
|
|
path := strings.TrimPrefix(r.URL.Path, "/api/v1/servers/")
|
|
parts := strings.Split(path, "/")
|
|
if path == "" || len(parts) < 2 || !controlPlaneResourceIDRE.MatchString(parts[0]) {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
clone := r.Clone(r.Context())
|
|
clone.URL.Path = "/v1/servers/" + path
|
|
s.serverMutation(w, clone)
|
|
}
|
|
|
|
type resultRequest struct {
|
|
MatchID string `json:"match_id"`
|
|
ResultNonce string `json:"result_nonce"`
|
|
Score struct {
|
|
Team0 int `json:"team_0"`
|
|
Team1 int `json:"team_1"`
|
|
} `json:"score"`
|
|
IntegrityState domain.IntegrityState `json:"integrity_state"`
|
|
}
|
|
|
|
type serverRegistrationRequest struct {
|
|
MatchID string `json:"match_id"`
|
|
ProtocolVersion int `json:"protocol_version"`
|
|
ImageDigest string `json:"image_digest"`
|
|
AssignmentReady bool `json:"assignment_ready"`
|
|
}
|
|
|
|
func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) {
|
|
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/servers/"), "/")
|
|
if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown" && parts[1] != "connect" && parts[1] != "disconnect") {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
if parts[1] == "roster" && r.Method != http.MethodGet || parts[1] != "roster" && r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) || ((parts[1] == "connect" || parts[1] == "disconnect") && s.ServerConnections == nil) {
|
|
writeError(w, http.StatusServiceUnavailable, "server_unavailable")
|
|
return
|
|
}
|
|
partsAuth := strings.Fields(r.Header.Get("Authorization"))
|
|
if len(partsAuth) != 2 || partsAuth[0] != "Bearer" || partsAuth[1] == "" {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return
|
|
}
|
|
now := s.now()
|
|
binding, err := s.WorkloadVerify(partsAuth[1], now)
|
|
if err != nil || binding.ServerID != parts[0] {
|
|
s.logEvent(observability.Event{Event: "server_" + parts[1], ServerID: parts[0], Stage: "unauthorized", OccurredAt: now})
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return
|
|
}
|
|
if parts[1] == "roster" {
|
|
roster, err := s.Roster(r.Context(), binding, now)
|
|
if err != nil || len(roster) == 0 {
|
|
writeError(w, http.StatusUnprocessableEntity, "roster_unavailable")
|
|
return
|
|
}
|
|
encodedRoster := make([]json.RawMessage, 0, len(roster))
|
|
for _, envelope := range roster {
|
|
encodedRoster = append(encodedRoster, json.RawMessage(envelope))
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(encodedRoster); err != nil {
|
|
return
|
|
}
|
|
return
|
|
}
|
|
key := r.Header.Get("Idempotency-Key")
|
|
if len(key) < 16 || len(key) > 128 {
|
|
writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
|
|
return
|
|
}
|
|
if parts[1] == "register" {
|
|
var input serverRegistrationRequest
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if input.MatchID == "" || input.MatchID != binding.MatchID || input.ProtocolVersion < 1 || !validImageDigest(input.ImageDigest) {
|
|
s.logEvent(observability.Event{Event: "server_register", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
if err := s.ServerRegistrar.RegisterServer(r.Context(), binding, input.ProtocolVersion, input.AssignmentReady, key, now); err != nil {
|
|
stage := "invalid"
|
|
if errors.Is(err, domain.ErrConflict) {
|
|
stage = "conflict"
|
|
s.Metrics.ObserveServerConflict("register")
|
|
writeError(w, http.StatusConflict, "conflict")
|
|
} else {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_register", MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now})
|
|
return
|
|
}
|
|
readyStage := "process_ready"
|
|
if input.AssignmentReady {
|
|
readyStage = "assignment_ready"
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_register", MatchID: binding.MatchID, ServerID: parts[0], Stage: readyStage, OccurredAt: now, Fields: map[string]any{"protocol_version": input.ProtocolVersion}})
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
if parts[1] == "connect" || parts[1] == "disconnect" {
|
|
var input struct {
|
|
PlayerID string `json:"player_id"`
|
|
Generation uint64 `json:"generation,omitempty"`
|
|
ExpectedGeneration *uint64 `json:"expected_generation,omitempty"`
|
|
}
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if !controlPlaneResourceIDRE.MatchString(input.PlayerID) {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
var generation uint64
|
|
var err error
|
|
if parts[1] == "connect" {
|
|
if input.Generation != 0 {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
expectedGeneration := uint64(0)
|
|
if input.ExpectedGeneration != nil {
|
|
expectedGeneration = *input.ExpectedGeneration
|
|
}
|
|
generation, err = s.ServerConnections.ClaimPlayerConnection(r.Context(), binding, input.PlayerID, expectedGeneration, key, now)
|
|
} else {
|
|
if input.Generation == 0 || input.ExpectedGeneration != nil {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
generation = input.Generation
|
|
err = s.ServerConnections.RecordPlayerDisconnected(r.Context(), binding, input.PlayerID, input.Generation, key, now)
|
|
}
|
|
if err != nil {
|
|
if errors.Is(err, domain.ErrConflict) {
|
|
s.Metrics.ObserveServerConflict(parts[1])
|
|
writeError(w, http.StatusConflict, "conflict")
|
|
} else {
|
|
// The request has already passed schema and workload checks. An
|
|
// unknown recorder error is infrastructure failure, not a terminal
|
|
// client fault; 503 keeps the game server's bounded retry alive.
|
|
writeError(w, http.StatusServiceUnavailable, "server_unavailable")
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_" + parts[1], MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
|
return
|
|
}
|
|
stage := "connected"
|
|
if parts[1] == "disconnect" {
|
|
stage = "disconnected"
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_" + parts[1], MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now, Fields: map[string]any{"player_id": input.PlayerID, "generation": generation}})
|
|
if parts[1] == "disconnect" {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
if input.ExpectedGeneration == nil {
|
|
// Rolling-upgrade compatibility for the pre-lease reporter. New
|
|
// servers always send expected_generation and consume the JSON lease.
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]uint64{"generation": generation})
|
|
return
|
|
}
|
|
if parts[1] == "shutdown" {
|
|
var input struct {
|
|
Reason string `json:"reason"`
|
|
}
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if input.Reason == "" || len(input.Reason) > 96 || strings.ContainsAny(input.Reason, "\r\n\t") {
|
|
s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
if err := s.ServerShutdowner.ShutdownServer(r.Context(), binding, input.Reason, key, now); err != nil {
|
|
stage := "invalid"
|
|
if errors.Is(err, domain.ErrConflict) {
|
|
stage = "conflict"
|
|
s.Metrics.ObserveServerConflict("shutdown")
|
|
writeError(w, http.StatusConflict, "conflict")
|
|
} else {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now})
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: "acknowledged", OccurredAt: now})
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
var input resultRequest
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
if input.MatchID == "" || binding.MatchID != input.MatchID || len(input.ResultNonce) < 16 || len(input.ResultNonce) > 128 || input.Score.Team0 < 0 || input.Score.Team1 < 0 || (input.IntegrityState != domain.IntegrityCertified && input.IntegrityState != domain.IntegritySuppressed && input.IntegrityState != domain.IntegrityReview) {
|
|
s.logEvent(observability.Event{Event: "server_result", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now})
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
return
|
|
}
|
|
result := domain.MatchResult{MatchID: input.MatchID, ServerID: parts[0], ResultNonce: input.ResultNonce, Team0Score: input.Score.Team0, Team1Score: input.Score.Team1, IntegrityState: input.IntegrityState}
|
|
payload, err := json.Marshal(input)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return
|
|
}
|
|
if err := s.ResultSubmitter.SubmitResult(r.Context(), key, result, binding, payload, now); err != nil {
|
|
stage := "invalid"
|
|
if errors.Is(err, domain.ErrResultConflict) || strings.Contains(err.Error(), "conflict") {
|
|
stage = "conflict"
|
|
s.Metrics.ObserveServerConflict("result")
|
|
writeError(w, http.StatusConflict, "conflict")
|
|
} else {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_result", MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now})
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "server_result", MatchID: binding.MatchID, ServerID: parts[0], Stage: "accepted", OccurredAt: now, Fields: map[string]any{"integrity_state": string(input.IntegrityState), "team_0": input.Score.Team0, "team_1": input.Score.Team1}})
|
|
w.WriteHeader(http.StatusAccepted)
|
|
}
|
|
|
|
func validImageDigest(value string) bool {
|
|
if len(value) != len("sha256:")+64 || !strings.HasPrefix(value, "sha256:") {
|
|
return false
|
|
}
|
|
for _, ch := range value[len("sha256:"):] {
|
|
if !(ch >= '0' && ch <= '9') && !(ch >= 'a' && ch <= 'f') {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost && r.Method != http.MethodGet && r.Method != http.MethodDelete {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
if s.Queue == nil && s.QueueBackend == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
|
|
return
|
|
}
|
|
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/queue/"), "/")
|
|
if r.Method == http.MethodGet {
|
|
if len(parts) != 1 || parts[0] == "" {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
var ticket domain.QueueTicket
|
|
var err error
|
|
now := s.now()
|
|
if s.QueueBackend != nil {
|
|
ticket, err = s.QueueBackend.Get(r.Context(), playerID, parts[0], now)
|
|
} else {
|
|
ticket, err = s.Queue.Get(playerID, parts[0], now)
|
|
}
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: "queue_get", QueueID: parts[0], Stage: "rejected", OccurredAt: now})
|
|
writeDomainError(w, err)
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "queue_get", QueueID: ticket.TicketID, Stage: strings.ToLower(string(ticket.State)), OccurredAt: now})
|
|
writeJSON(w, http.StatusOK, toQueueResponse(ticket))
|
|
return
|
|
}
|
|
if len(parts) != 2 || parts[0] == "" || (parts[1] != "heartbeat" && parts[1] != "cancel") {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
ticketID, key := parts[0], r.Header.Get("Idempotency-Key")
|
|
if len(key) < 16 || len(key) > 128 {
|
|
writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
|
|
return
|
|
}
|
|
revision, err := strconv.ParseUint(r.Header.Get("If-Match-Revision"), 10, 64)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_revision")
|
|
return
|
|
}
|
|
now := s.now()
|
|
var ticket domain.QueueTicket
|
|
if parts[1] == "heartbeat" {
|
|
if s.QueueBackend != nil {
|
|
ticket, err = s.QueueBackend.Heartbeat(r.Context(), playerID, ticketID, key, revision, now)
|
|
} else {
|
|
ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now)
|
|
}
|
|
} else {
|
|
if s.QueueBackend != nil {
|
|
ticket, err = s.QueueBackend.Cancel(r.Context(), playerID, ticketID, key, revision, now)
|
|
} else {
|
|
ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now)
|
|
}
|
|
}
|
|
eventName := "queue_heartbeat"
|
|
if parts[1] == "cancel" {
|
|
eventName = "queue_cancel"
|
|
}
|
|
if err != nil {
|
|
s.logQueueOutcome(eventName, ticketID, ticket, err, now)
|
|
writeDomainError(w, err)
|
|
return
|
|
}
|
|
s.logQueueOutcome(eventName, ticketID, ticket, nil, now)
|
|
if ticket.State == domain.Cancelled {
|
|
s.removeCandidate(r.Context(), ticket.TicketID)
|
|
} else {
|
|
s.projectCandidate(r.Context(), ticket)
|
|
}
|
|
s.publishTicketEvent(ticket, now)
|
|
if r.Header.Get("X-Contract-Delete") == "1" {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, toQueueResponse(ticket))
|
|
}
|
|
|
|
type proposalResponse struct {
|
|
ProposalID string `json:"proposal_id"`
|
|
Playlist string `json:"playlist"`
|
|
State string `json:"state"`
|
|
Revision uint64 `json:"revision"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
Participants []domain.ProposalParticipant `json:"participants"`
|
|
}
|
|
|
|
func (s *Service) proposalMutation(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost && r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/proposals/"), "/")
|
|
if r.Method == http.MethodGet {
|
|
if len(parts) != 1 || parts[0] == "" {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
s.proposalMu.Lock()
|
|
defer s.proposalMu.Unlock()
|
|
proposal, exists := s.Proposals[parts[0]]
|
|
if s.ProposalBackend != nil {
|
|
proposalValue, providerErr := s.ProposalBackend.Get(r.Context(), playerID, parts[0], s.now())
|
|
if providerErr != nil {
|
|
s.logEvent(observability.Event{Event: "proposal_get", ProposalID: parts[0], Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
proposal = &proposalValue
|
|
exists = true
|
|
}
|
|
if !exists || proposal == nil || !proposal.HasParticipant(playerID) {
|
|
s.logEvent(observability.Event{Event: "proposal_get", ProposalID: parts[0], Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
now := s.now()
|
|
if proposal.Expire(now) {
|
|
s.publishProposalEvent(*proposal, now)
|
|
}
|
|
s.logEvent(observability.Event{Event: "proposal_get", ProposalID: proposal.ProposalID, Stage: strings.ToLower(string(proposal.State)), OccurredAt: now})
|
|
writeJSON(w, http.StatusOK, toProposalResponse(*proposal))
|
|
return
|
|
}
|
|
if len(parts) != 2 || parts[0] == "" || (parts[1] != "accept" && parts[1] != "decline") {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
key := r.Header.Get("Idempotency-Key")
|
|
if len(key) < 16 || len(key) > 128 {
|
|
writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
|
|
return
|
|
}
|
|
revision, err := strconv.ParseUint(r.Header.Get("If-Match-Revision"), 10, 64)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_revision")
|
|
return
|
|
}
|
|
now := s.now()
|
|
var updated domain.Proposal
|
|
if s.ProposalBackend != nil {
|
|
mutator, supportsMutation := s.ProposalBackend.(ProposalMutationBackend)
|
|
if !supportsMutation {
|
|
writeError(w, http.StatusServiceUnavailable, "proposal_unavailable")
|
|
return
|
|
}
|
|
updated, err = mutator.Respond(r.Context(), playerID, parts[0], key, parts[1] == "accept", revision, now)
|
|
} else {
|
|
s.proposalMu.Lock()
|
|
defer s.proposalMu.Unlock()
|
|
proposal, exists := s.Proposals[parts[0]]
|
|
if !exists || proposal == nil {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
updated, err = proposal.Respond(playerID, key, parts[1] == "accept", revision, now)
|
|
}
|
|
if err != nil {
|
|
s.logProposalOutcome(parts[0], updated, err, now)
|
|
writeDomainError(w, err)
|
|
return
|
|
}
|
|
s.logProposalOutcome(parts[0], updated, nil, now)
|
|
if updated.State == domain.Accepted && s.ProposalPromoter != nil {
|
|
if err := s.ProposalPromoter.Promote(r.Context(), updated, now); err != nil {
|
|
writeError(w, http.StatusServiceUnavailable, "match_promotion_unavailable")
|
|
return
|
|
}
|
|
}
|
|
s.publishProposalEvent(updated, now)
|
|
writeJSON(w, http.StatusOK, toProposalResponse(updated))
|
|
}
|
|
|
|
func (s *Service) assignment(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/assignments/"), "/")
|
|
if len(parts) != 1 || parts[0] == "" {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
if s.Assignment == nil {
|
|
s.logEvent(observability.Event{Event: "assignment_get", MatchID: parts[0], Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusServiceUnavailable, "assignment_unavailable")
|
|
return
|
|
}
|
|
now := s.now()
|
|
view, err := s.Assignment(r.Context(), playerID, parts[0], now)
|
|
if err != nil || view.MatchID != parts[0] || view.PlayerID != playerID {
|
|
s.logEvent(observability.Event{Event: "assignment_get", MatchID: parts[0], ServerID: view.ServerID, Stage: "rejected", OccurredAt: now})
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
if view.ServerID == "" || view.Slot < 0 || view.Slot > 5 || view.ProtocolVersion < 1 || (view.Transport != "enet" && view.Transport != "steam_sdr") || view.JoinAuthorisation == "" || !validAssignmentEndpoint(view.Endpoint) || view.ExpiresAt.IsZero() || !now.Before(view.ExpiresAt) {
|
|
s.logEvent(observability.Event{Event: "assignment_get", MatchID: parts[0], ServerID: view.ServerID, Stage: "rejected", OccurredAt: now})
|
|
writeError(w, http.StatusServiceUnavailable, "assignment_unavailable")
|
|
return
|
|
}
|
|
_ = s.PublishControlPlaneEvent(assignmentChangedEvent(view, now))
|
|
s.logEvent(observability.Event{Event: "assignment_get", MatchID: view.MatchID, ServerID: view.ServerID, Stage: "assignment_ready", OccurredAt: now})
|
|
writeJSON(w, http.StatusOK, view)
|
|
}
|
|
|
|
func validAssignmentEndpoint(endpoint string) bool {
|
|
if endpoint == "" || strings.ContainsAny(endpoint, "/?#") {
|
|
return false
|
|
}
|
|
host, portText, err := net.SplitHostPort(endpoint)
|
|
if err != nil || host == "" {
|
|
return false
|
|
}
|
|
port, err := strconv.Atoi(portText)
|
|
return err == nil && port >= 1 && port <= 65535
|
|
}
|
|
|
|
func assignmentChangedEvent(view AssignmentView, now time.Time) ControlPlaneEvent {
|
|
return ControlPlaneEvent{Event: "assignment_changed", Revision: view.Revision, ResourceID: view.MatchID, OccurredAt: now, MatchID: view.MatchID, ServerID: view.ServerID, PlayerID: view.PlayerID}
|
|
}
|
|
|
|
type rankedProfileResponse struct {
|
|
Rating float64 `json:"rating"`
|
|
RD float64 `json:"rd"`
|
|
Volatility float64 `json:"volatility"`
|
|
RankedGames int `json:"ranked_games"`
|
|
Tier string `json:"tier"`
|
|
Provisional bool `json:"provisional"`
|
|
SeasonID string `json:"season_id,omitempty"`
|
|
SeasonEndsAt string `json:"season_ends_at,omitempty"`
|
|
}
|
|
|
|
func (s *Service) profile(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
profile, exists, err := s.rankedProfileFor(r.Context(), playerID)
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: "profile_get", Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusServiceUnavailable, "ranked_profile_unavailable")
|
|
return
|
|
}
|
|
if !exists {
|
|
s.logEvent(observability.Event{Event: "profile_get", Stage: "not_found", OccurredAt: s.now()})
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "profile_get", Stage: "ok", OccurredAt: s.now()})
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"player_id": playerID,
|
|
"rating": profile.Value,
|
|
"rd": profile.RD,
|
|
"provisional": domain.RankedIsProvisional(profile),
|
|
})
|
|
}
|
|
|
|
func (s *Service) rankedProfile(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
profile, exists, err := s.rankedProfileFor(r.Context(), playerID)
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: "ranked_profile_get", Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusServiceUnavailable, "ranked_profile_unavailable")
|
|
return
|
|
}
|
|
if !exists {
|
|
s.logEvent(observability.Event{Event: "ranked_profile_get", Stage: "not_found", OccurredAt: s.now()})
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
tier, err := domain.RankedTier(profile, s.TierPolicy)
|
|
if err != nil {
|
|
s.logEvent(observability.Event{Event: "ranked_profile_get", Stage: "rejected", OccurredAt: s.now()})
|
|
writeError(w, http.StatusServiceUnavailable, "ranked_profile_unavailable")
|
|
return
|
|
}
|
|
s.logEvent(observability.Event{Event: "ranked_profile_get", Stage: "ok", OccurredAt: s.now()})
|
|
seasonID := profile.CurrentSeasonID
|
|
if seasonID == "" {
|
|
seasonID = profile.LastSeasonID
|
|
}
|
|
seasonEndsAt := ""
|
|
if profile.CurrentSeasonID != "" && !profile.CurrentSeasonEndsAt.IsZero() {
|
|
seasonEndsAt = profile.CurrentSeasonEndsAt.UTC().Format(time.RFC3339)
|
|
}
|
|
writeJSON(w, http.StatusOK, rankedProfileResponse{Rating: profile.Value, RD: profile.RD, Volatility: profile.Volatility, RankedGames: profile.RankedGames, Tier: string(tier), Provisional: domain.RankedIsProvisional(profile), SeasonID: seasonID, SeasonEndsAt: seasonEndsAt})
|
|
}
|
|
|
|
type probeRequest struct {
|
|
OpaqueLocation []byte `json:"opaque_location"`
|
|
Nonce []byte `json:"nonce"`
|
|
}
|
|
|
|
func (s *Service) probe(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
|
|
return
|
|
}
|
|
playerID, ok := s.authenticate(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
region := strings.TrimPrefix(r.URL.Path, "/v1/probes/")
|
|
if (region != "EU" && region != "NA") || strings.Contains(region, "/") {
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
return
|
|
}
|
|
if s.Probe == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "probe_unavailable")
|
|
return
|
|
}
|
|
var input probeRequest
|
|
if !decodeBody(w, r, &input) {
|
|
return
|
|
}
|
|
receivedAt := s.now()
|
|
evidence, expectedNonce, err := s.Probe(playerID, region, input.OpaqueLocation, input.Nonce, receivedAt)
|
|
if err != nil {
|
|
writeError(w, http.StatusUnprocessableEntity, "probe_unavailable")
|
|
return
|
|
}
|
|
if evidence.Region != region || domain.ValidateProbe(evidence, expectedNonce, receivedAt) != nil {
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_probe")
|
|
return
|
|
}
|
|
if s.ProbeRecorder != nil {
|
|
if err := s.ProbeRecorder.RecordProbe(r.Context(), playerID, region, evidence.ServerRTT, receivedAt); err != nil {
|
|
writeError(w, http.StatusServiceUnavailable, "probe_persistence_failed")
|
|
return
|
|
}
|
|
}
|
|
writeJSON(w, http.StatusAccepted, map[string]any{"region": region, "server_rtt_ms": evidence.ServerRTT.Milliseconds(), "status": "accepted"})
|
|
}
|
|
|
|
func (s *Service) authenticate(w http.ResponseWriter, r *http.Request) (string, bool) {
|
|
if s.Sessions == nil && s.SessionBackend == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "auth_unavailable")
|
|
return "", false
|
|
}
|
|
parts := strings.Fields(r.Header.Get("Authorization"))
|
|
if len(parts) != 2 || parts[0] != "Bearer" {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return "", false
|
|
}
|
|
separator := strings.IndexByte(parts[1], ':')
|
|
if separator <= 0 || separator == len(parts[1])-1 {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return "", false
|
|
}
|
|
var session domain.Session
|
|
var err error
|
|
if s.SessionBackend != nil {
|
|
session, err = s.SessionBackend.Authenticate(r.Context(), parts[1][:separator], parts[1][separator+1:], s.now())
|
|
} else {
|
|
session, err = s.Sessions.Authenticate(parts[1][:separator], parts[1][separator+1:], s.now())
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return "", false
|
|
}
|
|
return session.PlayerID, true
|
|
}
|
|
|
|
func (s *Service) now() time.Time {
|
|
if s.Now != nil {
|
|
return s.Now()
|
|
}
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
func decodeBody(w http.ResponseWriter, r *http.Request, target any) bool {
|
|
r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
|
|
decoder := json.NewDecoder(r.Body)
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(target); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return false
|
|
}
|
|
var extra any
|
|
if err := decoder.Decode(&extra); err != io.EOF {
|
|
writeError(w, http.StatusBadRequest, "invalid_request")
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func toQueueResponse(ticket domain.QueueTicket) queueResponse {
|
|
return queueResponse{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), Revision: ticket.Revision, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt}
|
|
}
|
|
|
|
func toProposalResponse(proposal domain.Proposal) proposalResponse {
|
|
return proposalResponse{ProposalID: proposal.ProposalID, Playlist: string(proposal.Playlist), State: string(proposal.State), Revision: proposal.Revision, ExpiresAt: proposal.ExpiresAt, Participants: proposal.Participants}
|
|
}
|
|
|
|
func writeDomainError(w http.ResponseWriter, err error) {
|
|
switch {
|
|
case errors.Is(err, domain.ErrPlayerQueued), errors.Is(err, domain.ErrConflict), errors.Is(err, domain.ErrStaleRevision):
|
|
writeError(w, http.StatusConflict, "conflict")
|
|
case errors.Is(err, domain.ErrPlayerCooldown):
|
|
writeError(w, http.StatusTooManyRequests, "matchmaking_cooldown")
|
|
case errors.Is(err, domain.ErrTicketExpired):
|
|
writeError(w, http.StatusGone, "expired")
|
|
case errors.Is(err, domain.ErrNotTicketOwner):
|
|
writeError(w, http.StatusForbidden, "forbidden")
|
|
case errors.Is(err, domain.ErrTicketNotFound):
|
|
writeError(w, http.StatusNotFound, "not_found")
|
|
default:
|
|
writeError(w, http.StatusUnprocessableEntity, "invalid_request")
|
|
}
|
|
}
|
|
|
|
func writeError(w http.ResponseWriter, status int, code string) {
|
|
writeJSON(w, status, map[string]string{"error": code})
|
|
}
|
|
|
|
func writeJSON(w http.ResponseWriter, status int, value any) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(status)
|
|
_ = json.NewEncoder(w).Encode(value)
|
|
}
|