mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
325 lines
16 KiB
Go
325 lines
16 KiB
Go
package store
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
|
)
|
|
|
|
const (
|
|
QueueIdempotencyScope = "queue.create"
|
|
QueueIdempotencyInsertSQL = `INSERT INTO idempotency_keys (scope, idempotency_key, payload_digest, result)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (scope, idempotency_key) DO NOTHING`
|
|
QueueIdempotencySelectSQL = `SELECT payload_digest, result
|
|
FROM idempotency_keys
|
|
WHERE scope = $1 AND idempotency_key = $2
|
|
FOR UPDATE`
|
|
QueueTicketSelectSQL = `SELECT q.ticket_id, q.player_id, q.playlist, q.state, q.client_build,
|
|
q.protocol_version, q.enqueued_at, q.expires_at, q.revision, q.predicted_rtt,
|
|
COALESCE((SELECT pp.proposal_id FROM proposal_participants pp
|
|
JOIN proposals p ON p.proposal_id = pp.proposal_id
|
|
WHERE pp.ticket_id = q.ticket_id AND pp.player_id = q.player_id
|
|
AND p.state = 'OPEN'
|
|
LIMIT 1), ''),
|
|
COALESCE((SELECT mp.match_id FROM match_participants mp
|
|
WHERE mp.ticket_id = q.ticket_id AND mp.player_id = q.player_id
|
|
AND mp.participation_active
|
|
LIMIT 1), '')
|
|
FROM queue_tickets q
|
|
WHERE q.ticket_id = $1 AND q.player_id = $2`
|
|
QueueTicketHeartbeatSQL = `UPDATE queue_tickets SET revision = revision + 1,
|
|
expires_at = $4 + INTERVAL '30 seconds'
|
|
WHERE ticket_id = $1 AND player_id = $2 AND revision = $3
|
|
AND state IN ('QUEUED', 'PROPOSED') AND expires_at > $4
|
|
RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision, predicted_rtt`
|
|
QueueTicketCancelSQL = `UPDATE queue_tickets SET state = 'CANCELLED',
|
|
revision = revision + 1, expires_at = $4
|
|
WHERE ticket_id = $1 AND player_id = $2 AND revision = $3
|
|
AND state NOT IN ('COMPLETED', 'CANCELLED', 'EXPIRED', 'FAILED')
|
|
RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision, predicted_rtt`
|
|
QueueCooldownSelectSQL = `SELECT ends_at
|
|
FROM penalties
|
|
WHERE player_id = $1 AND playlist = $2
|
|
AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT')
|
|
AND ends_at > $3
|
|
ORDER BY ends_at DESC
|
|
LIMIT 1`
|
|
)
|
|
|
|
const QueueCandidateProjectionSQL = `SELECT ticket_id, player_id, playlist, client_build,
|
|
protocol_version, enqueued_at, predicted_rtt
|
|
FROM queue_tickets
|
|
WHERE state = 'QUEUED' AND playlist = $1 AND expires_at > $2
|
|
ORDER BY enqueued_at, ticket_id
|
|
LIMIT $3`
|
|
|
|
const RankedParticipantSQL = `SELECT player_id, steam_id
|
|
FROM identities
|
|
WHERE player_id = ANY($1)
|
|
ORDER BY player_id`
|
|
|
|
// LoadRankedParticipants resolves the verified identity metadata required by
|
|
// ranked admission. The caller must compare the returned set with the formed
|
|
// candidate set; a partial lookup is not a valid ranked roster.
|
|
func LoadRankedParticipants(ctx context.Context, db *sql.DB, playerIDs []string) ([]domain.RankedParticipant, error) {
|
|
if db == nil || len(playerIDs) != 6 {
|
|
return nil, fmt.Errorf("ranked admission requires six players")
|
|
}
|
|
rows, err := db.QueryContext(ctx, RankedParticipantSQL, playerIDs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
participants := make([]domain.RankedParticipant, 0, len(playerIDs))
|
|
for rows.Next() {
|
|
var participant domain.RankedParticipant
|
|
if err := rows.Scan(&participant.PlayerID, &participant.SteamID); err != nil {
|
|
return nil, err
|
|
}
|
|
participants = append(participants, participant)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(participants) != len(playerIDs) {
|
|
return nil, fmt.Errorf("ranked identity metadata is incomplete")
|
|
}
|
|
return participants, nil
|
|
}
|
|
|
|
// ListQueuedCandidates is an authoritative, expiry-filtered source for the
|
|
// matcher projection. It deliberately does not claim rows; CreateProposal is
|
|
// the transaction that performs the competing claim with SKIP LOCKED fences.
|
|
func ListQueuedCandidates(ctx context.Context, db *sql.DB, playlist domain.Playlist, now time.Time, limit int) ([]domain.Candidate, error) {
|
|
if db == nil || (playlist != domain.Casual && playlist != domain.Ranked) || now.IsZero() || limit < 1 || limit > 1000 {
|
|
return nil, fmt.Errorf("invalid queued candidate arguments")
|
|
}
|
|
rows, err := db.QueryContext(ctx, QueueCandidateProjectionSQL, string(playlist), now, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var candidates []domain.Candidate
|
|
for rows.Next() {
|
|
var candidate domain.Candidate
|
|
var playlist string
|
|
var predictedRTT []byte
|
|
if err := rows.Scan(&candidate.TicketID, &candidate.PlayerID, &playlist, &candidate.ClientBuild, &candidate.ProtocolVersion, &candidate.EnqueuedAt, &predictedRTT); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := json.Unmarshal(predictedRTT, &candidate.PredictedRTT); err != nil {
|
|
return nil, fmt.Errorf("decode candidate RTT: %w", err)
|
|
}
|
|
candidate.Playlist = domain.Playlist(playlist)
|
|
candidates = append(candidates, candidate)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return candidates, nil
|
|
}
|
|
|
|
func CreateQueueTicket(ctx context.Context, db *sql.DB, ticketID, playerID, idempotencyKey string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) {
|
|
if db == nil || ticketID == "" || playerID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || (spec.Playlist != domain.Casual && spec.Playlist != domain.Ranked) || spec.ClientBuild == "" || len(spec.ClientBuild) > 128 || spec.ProtocolVersion < 1 || now.IsZero() {
|
|
return domain.QueueTicket{}, fmt.Errorf("invalid queue transaction arguments")
|
|
}
|
|
digest := sha256.Sum256([]byte(fmt.Sprintf("%s|%s|%s|%s|%d", ticketID, playerID, spec.Playlist, spec.ClientBuild, spec.ProtocolVersion)))
|
|
var ticket domain.QueueTicket
|
|
err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
|
candidate := domain.Candidate{TicketID: ticketID, PlayerID: playerID, Playlist: spec.Playlist, ClientBuild: spec.ClientBuild, ProtocolVersion: spec.ProtocolVersion, EnqueuedAt: now}
|
|
ticket = domain.QueueTicket{TicketID: ticketID, PlayerID: playerID, Candidate: candidate, Playlist: spec.Playlist, State: domain.Queued, EnqueuedAt: now, ExpiresAt: now.Add(domain.QueueExpiryWindow)}
|
|
stored, err := json.Marshal(queueTicketRecordFromDomain(ticket))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result, err := tx.ExecContext(ctx, QueueIdempotencyInsertSQL, QueueIdempotencyScope, idempotencyKey, digest[:], stored)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
inserted, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if inserted == 0 {
|
|
var priorDigest, priorResult []byte
|
|
if err := tx.QueryRowContext(ctx, QueueIdempotencySelectSQL, QueueIdempotencyScope, idempotencyKey).Scan(&priorDigest, &priorResult); err != nil {
|
|
return err
|
|
}
|
|
if !bytes.Equal(priorDigest, digest[:]) {
|
|
return fmt.Errorf("queue create idempotency conflict")
|
|
}
|
|
var prior queueTicketRecord
|
|
if err := json.Unmarshal(priorResult, &prior); err != nil {
|
|
return fmt.Errorf("invalid stored queue result: %w", err)
|
|
}
|
|
ticket = queueTicketRecordToDomain(prior)
|
|
return nil
|
|
}
|
|
var cooldownEndsAt time.Time
|
|
if err := tx.QueryRowContext(ctx, QueueCooldownSelectSQL, playerID, string(spec.Playlist), now).Scan(&cooldownEndsAt); err != sql.ErrNoRows {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return fmt.Errorf("%w until %s", domain.ErrPlayerCooldown, cooldownEndsAt.UTC().Format(time.RFC3339))
|
|
}
|
|
predictedRTT, err := json.Marshal(candidate.PredictedRTT)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, QueueTicketInsertSQL, ticketID, playerID, string(spec.Playlist), spec.ClientBuild, spec.ProtocolVersion, now, ticket.ExpiresAt, predictedRTT)
|
|
return err
|
|
})
|
|
return ticket, err
|
|
}
|
|
|
|
type queueTicketRecord struct {
|
|
TicketID string `json:"ticket_id"`
|
|
PlayerID string `json:"player_id"`
|
|
ProposalID string `json:"proposal_id,omitempty"`
|
|
MatchID string `json:"match_id,omitempty"`
|
|
Playlist string `json:"playlist"`
|
|
State string `json:"state"`
|
|
ClientBuild string `json:"client_build"`
|
|
ProtocolVersion int `json:"protocol_version"`
|
|
EnqueuedAt time.Time `json:"enqueued_at"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
Revision uint64 `json:"revision"`
|
|
PredictedRTT map[string]float64 `json:"predicted_rtt"`
|
|
}
|
|
|
|
type PostgresQueue struct{ DB *sql.DB }
|
|
|
|
func (q PostgresQueue) RecordProviderAllocation(ctx context.Context, allocation domain.Allocation, now time.Time) (domain.Allocation, error) {
|
|
return RecordProviderAllocation(ctx, q.DB, allocation, now)
|
|
}
|
|
|
|
const QueueProbeRecordSQL = `UPDATE queue_tickets
|
|
SET predicted_rtt = jsonb_set(COALESCE(predicted_rtt, '{}'::jsonb), ARRAY[$2], to_jsonb($3::double precision), true)
|
|
WHERE player_id = $1 AND state IN ('QUEUED', 'PROPOSED') AND expires_at > $4`
|
|
|
|
func (q PostgresQueue) RecordProbe(ctx context.Context, playerID, region string, rtt time.Duration, now time.Time) error {
|
|
if q.DB == nil || playerID == "" || (region != "EU" && region != "NA") || rtt < 0 || now.IsZero() {
|
|
return fmt.Errorf("invalid probe recording")
|
|
}
|
|
result, err := q.DB.ExecContext(ctx, QueueProbeRecordSQL, playerID, region, float64(rtt)/float64(time.Millisecond), now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
changed, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if changed == 0 {
|
|
return domain.ErrTicketNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (q PostgresQueue) Create(ctx context.Context, playerID, ticketID, idempotencyKey string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) {
|
|
return CreateQueueTicket(ctx, q.DB, ticketID, playerID, idempotencyKey, spec, now)
|
|
}
|
|
func (q PostgresQueue) Heartbeat(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) {
|
|
return HeartbeatQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now)
|
|
}
|
|
func (q PostgresQueue) Cancel(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) {
|
|
return CancelQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now)
|
|
}
|
|
func (q PostgresQueue) Get(ctx context.Context, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) {
|
|
return GetQueueTicket(ctx, q.DB, playerID, ticketID, now)
|
|
}
|
|
|
|
func GetQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) {
|
|
if db == nil || playerID == "" || ticketID == "" || now.IsZero() {
|
|
return domain.QueueTicket{}, fmt.Errorf("invalid queue recovery arguments")
|
|
}
|
|
var record queueTicketRecord
|
|
var predictedRTT []byte
|
|
if err := db.QueryRowContext(ctx, QueueTicketSelectSQL, ticketID, playerID).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT, &record.ProposalID, &record.MatchID); err != nil {
|
|
return domain.QueueTicket{}, err
|
|
}
|
|
if err := json.Unmarshal(predictedRTT, &record.PredictedRTT); err != nil {
|
|
return domain.QueueTicket{}, fmt.Errorf("decode queue RTT: %w", err)
|
|
}
|
|
ticket := queueTicketRecordToDomain(record)
|
|
if (ticket.State == domain.Queued || ticket.State == domain.Proposed) && !now.Before(ticket.ExpiresAt) {
|
|
return domain.QueueTicket{}, domain.ErrTicketExpired
|
|
}
|
|
return ticket, nil
|
|
}
|
|
|
|
func HeartbeatQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idempotencyKey string, expectedRevision uint64, now time.Time) (domain.QueueTicket, error) {
|
|
return mutateQueueTicket(ctx, db, playerID, ticketID, idempotencyKey, expectedRevision, now, "heartbeat", QueueTicketHeartbeatSQL)
|
|
}
|
|
|
|
func CancelQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idempotencyKey string, expectedRevision uint64, now time.Time) (domain.QueueTicket, error) {
|
|
return mutateQueueTicket(ctx, db, playerID, ticketID, idempotencyKey, expectedRevision, now, "cancel", QueueTicketCancelSQL)
|
|
}
|
|
|
|
func mutateQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID, idempotencyKey string, expectedRevision uint64, now time.Time, operation, mutationSQL string) (ticket domain.QueueTicket, err error) {
|
|
if db == nil || playerID == "" || ticketID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() || (operation != "heartbeat" && operation != "cancel") {
|
|
return domain.QueueTicket{}, fmt.Errorf("invalid queue mutation arguments")
|
|
}
|
|
digest := sha256.Sum256([]byte(fmt.Sprintf("%s|%s|%s|%d", operation, playerID, ticketID, expectedRevision)))
|
|
err = RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
|
result, err := tx.ExecContext(ctx, QueueIdempotencyInsertSQL, QueueIdempotencyScope+"."+operation, idempotencyKey, digest[:], []byte("{}"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
inserted, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if inserted == 0 {
|
|
var priorDigest, priorResult []byte
|
|
if err := tx.QueryRowContext(ctx, QueueIdempotencySelectSQL, QueueIdempotencyScope+"."+operation, idempotencyKey).Scan(&priorDigest, &priorResult); err != nil {
|
|
return err
|
|
}
|
|
if !bytes.Equal(priorDigest, digest[:]) {
|
|
return fmt.Errorf("queue mutation idempotency conflict")
|
|
}
|
|
var prior queueTicketRecord
|
|
if err := json.Unmarshal(priorResult, &prior); err != nil {
|
|
return fmt.Errorf("invalid stored queue result: %w", err)
|
|
}
|
|
ticket = queueTicketRecordToDomain(prior)
|
|
return nil
|
|
}
|
|
var record queueTicketRecord
|
|
var predictedRTT []byte
|
|
if err := tx.QueryRowContext(ctx, mutationSQL, ticketID, playerID, expectedRevision, now).Scan(&record.TicketID, &record.PlayerID, &record.Playlist, &record.State, &record.ClientBuild, &record.ProtocolVersion, &record.EnqueuedAt, &record.ExpiresAt, &record.Revision, &predictedRTT); err != nil {
|
|
return fmt.Errorf("queue mutation rejected: %w", err)
|
|
}
|
|
if err := json.Unmarshal(predictedRTT, &record.PredictedRTT); err != nil {
|
|
return fmt.Errorf("decode queue RTT: %w", err)
|
|
}
|
|
ticket = queueTicketRecordToDomain(record)
|
|
if operation == "cancel" {
|
|
if err := CascadeCancelToOpenProposal(ctx, tx, ticketID, playerID, now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
stored, err := json.Marshal(record)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, QueueIdempotencyScope+"."+operation, idempotencyKey, stored)
|
|
return err
|
|
})
|
|
return ticket, err
|
|
}
|
|
|
|
func queueTicketRecordFromDomain(ticket domain.QueueTicket) queueTicketRecord {
|
|
return queueTicketRecord{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, ProposalID: ticket.ProposalID, MatchID: ticket.MatchID, Playlist: string(ticket.Playlist), State: string(ticket.State), ClientBuild: ticket.Candidate.ClientBuild, ProtocolVersion: ticket.Candidate.ProtocolVersion, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt, Revision: ticket.Revision, PredictedRTT: ticket.Candidate.PredictedRTT}
|
|
}
|
|
func queueTicketRecordToDomain(record queueTicketRecord) domain.QueueTicket {
|
|
candidate := domain.Candidate{TicketID: record.TicketID, PlayerID: record.PlayerID, Playlist: domain.Playlist(record.Playlist), ClientBuild: record.ClientBuild, ProtocolVersion: record.ProtocolVersion, EnqueuedAt: record.EnqueuedAt, PredictedRTT: record.PredictedRTT}
|
|
return domain.QueueTicket{TicketID: record.TicketID, PlayerID: record.PlayerID, ProposalID: record.ProposalID, MatchID: record.MatchID, Candidate: candidate, Playlist: domain.Playlist(record.Playlist), State: domain.State(record.State), Revision: record.Revision, EnqueuedAt: record.EnqueuedAt, ExpiresAt: record.ExpiresAt}
|
|
}
|