mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 08:23:45 +00:00
feat(multiplayer): apply initial connect outcomes durably
This commit is contained in:
Josh Creek
@@ -0,0 +1,252 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
||||
)
|
||||
|
||||
const InitialConnectIdempotencyScope = "match.initial_connect"
|
||||
|
||||
const initialConnectMatchLockSQL = `SELECT playlist, state, revision
|
||||
FROM matches WHERE match_id = $1 FOR UPDATE`
|
||||
|
||||
const initialConnectParticipantsSQL = `SELECT player_id, ticket_id, team, connected_at,
|
||||
participation_active
|
||||
FROM match_participants WHERE match_id = $1 ORDER BY player_id FOR UPDATE`
|
||||
|
||||
const initialConnectIdempotencyInsertSQL = `INSERT INTO idempotency_keys
|
||||
(scope, idempotency_key, payload_digest, result)
|
||||
VALUES ($1, $2, $3, '{}'::jsonb) ON CONFLICT (scope, idempotency_key) DO NOTHING`
|
||||
|
||||
const initialConnectIdempotencySelectSQL = `SELECT payload_digest, result
|
||||
FROM idempotency_keys WHERE scope = $1 AND idempotency_key = $2 FOR UPDATE`
|
||||
|
||||
const initialConnectMatchUpdateSQL = `UPDATE matches
|
||||
SET state = $2, revision = revision + 1 WHERE match_id = $1
|
||||
RETURNING revision`
|
||||
|
||||
const initialConnectDeactivateSQL = `UPDATE match_participants
|
||||
SET participation_active = FALSE, abandoned_at = $3
|
||||
WHERE match_id = $1 AND player_id = ANY($2)`
|
||||
|
||||
const initialConnectTicketNoShowSQL = `UPDATE queue_tickets q
|
||||
SET state = 'FAILED', revision = revision + 1
|
||||
FROM match_participants mp
|
||||
WHERE mp.match_id = $1 AND mp.player_id = ANY($2)
|
||||
AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
|
||||
AND q.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')`
|
||||
|
||||
const initialConnectTicketInnocentCancelSQL = `UPDATE queue_tickets q
|
||||
SET state = 'QUEUED', expires_at = $2, revision = revision + 1
|
||||
FROM match_participants mp
|
||||
WHERE mp.match_id = $1 AND mp.player_id = ANY($3)
|
||||
AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
|
||||
AND q.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')`
|
||||
|
||||
const initialConnectTicketConnectedLiveSQL = `UPDATE queue_tickets q
|
||||
SET state = 'LIVE', revision = revision + 1
|
||||
FROM match_participants mp
|
||||
WHERE mp.match_id = $1 AND mp.player_id = ANY($2)
|
||||
AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
|
||||
AND q.state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')`
|
||||
|
||||
const initialConnectPenaltySQL = `INSERT INTO penalties
|
||||
(penalty_id, player_id, match_id, playlist, kind, starts_at, ends_at)
|
||||
VALUES ($1, $2, $3, $4, 'INITIAL_CONNECT_NO_SHOW', $5, $6)
|
||||
ON CONFLICT (penalty_id) DO NOTHING`
|
||||
|
||||
const initialConnectOutboxSQL = `INSERT INTO outbox
|
||||
(event_id, aggregate_type, aggregate_id, revision, event_type, payload)
|
||||
VALUES ($1, 'match', $2, $3, 'state_changed', $4)
|
||||
ON CONFLICT DO NOTHING`
|
||||
|
||||
type initialConnectParticipant struct {
|
||||
PlayerID string
|
||||
TicketID string
|
||||
Team int
|
||||
ConnectedAt sql.NullTime
|
||||
Active bool
|
||||
}
|
||||
|
||||
// ApplyInitialConnectPlan atomically reconciles the pre-live connect window.
|
||||
// It is deliberately a store operation: no-show penalties and innocent-ticket
|
||||
// requeue must commit with the match transition or neither may commit.
|
||||
func ApplyInitialConnectPlan(ctx context.Context, db *sql.DB, matchID, idempotencyKey string, plan domain.InitialConnectPlan, now time.Time) error {
|
||||
if db == nil || matchID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() || plan.Action == domain.InitialConnectWait || (plan.Action != domain.InitialConnectCancel && plan.Action != domain.InitialConnectStartWithBot) || plan.MatchState == domain.Live && plan.Action != domain.InitialConnectStartWithBot || plan.MatchState == domain.Cancelled && plan.Action != domain.InitialConnectCancel {
|
||||
return fmt.Errorf("invalid initial-connect transaction arguments")
|
||||
}
|
||||
digest, err := initialConnectDigest(matchID, plan)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
|
||||
inserted, err := tx.ExecContext(ctx, initialConnectIdempotencyInsertSQL, InitialConnectIdempotencyScope, idempotencyKey, digest[:])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
count, err := inserted.RowsAffected()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if count == 0 {
|
||||
var prior []byte
|
||||
var result []byte
|
||||
if err := tx.QueryRowContext(ctx, initialConnectIdempotencySelectSQL, InitialConnectIdempotencyScope, idempotencyKey).Scan(&prior, &result); err != nil {
|
||||
return err
|
||||
}
|
||||
if !bytes.Equal(prior, digest[:]) {
|
||||
return fmt.Errorf("conflicting initial-connect request")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var playlist, state string
|
||||
var revision int64
|
||||
if err := tx.QueryRowContext(ctx, initialConnectMatchLockSQL, matchID).Scan(&playlist, &state, &revision); err != nil {
|
||||
return err
|
||||
}
|
||||
if state != string(domain.AssignmentReady) && state != string(domain.Assigned) && state != string(domain.Connecting) {
|
||||
return fmt.Errorf("match is not awaiting initial connect: %s", state)
|
||||
}
|
||||
participants, err := loadInitialConnectParticipants(ctx, tx, matchID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateInitialConnectPlan(plan, participants, domain.Playlist(playlist)); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, initialConnectDeactivateSQL, matchID, initialConnectNoShowIDs(plan), now); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, initialConnectTicketNoShowSQL, matchID, initialConnectNoShowIDs(plan)); err != nil {
|
||||
return err
|
||||
}
|
||||
if plan.Action == domain.InitialConnectCancel {
|
||||
if _, err := tx.ExecContext(ctx, initialConnectTicketInnocentCancelSQL, matchID, now.Add(domain.QueueExpiryWindow), plan.Connected); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if _, err := tx.ExecContext(ctx, initialConnectTicketConnectedLiveSQL, matchID, plan.Connected); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, noShow := range plan.NoShows {
|
||||
penaltyID := "initial-connect:" + matchID + ":" + noShow.PlayerID
|
||||
if _, err := tx.ExecContext(ctx, initialConnectPenaltySQL, penaltyID, noShow.PlayerID, matchID, playlist, noShow.AbandonedAt, noShow.AbandonedAt.Add(noShow.Cooldown)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
var finalRevision int64
|
||||
if err := tx.QueryRowContext(ctx, initialConnectMatchUpdateSQL, matchID, string(plan.MatchState)).Scan(&finalRevision); err != nil {
|
||||
return err
|
||||
}
|
||||
payload, _ := json.Marshal(map[string]any{"match_id": matchID, "state": plan.MatchState, "action": plan.Action})
|
||||
if _, err := tx.ExecContext(ctx, initialConnectOutboxSQL, "initial-connect:"+matchID+fmt.Sprintf(":%d", finalRevision), matchID, finalRevision, payload); err != nil {
|
||||
return err
|
||||
}
|
||||
stored, _ := json.Marshal(map[string]any{"match_id": matchID, "state": plan.MatchState, "revision": finalRevision})
|
||||
_, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, InitialConnectIdempotencyScope, idempotencyKey, stored)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func initialConnectNoShowIDs(plan domain.InitialConnectPlan) []string {
|
||||
result := make([]string, len(plan.NoShows))
|
||||
for i := range plan.NoShows {
|
||||
result[i] = plan.NoShows[i].PlayerID
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func initialConnectDigest(matchID string, plan domain.InitialConnectPlan) ([32]byte, error) {
|
||||
copyPlan := plan
|
||||
sort.Strings(copyPlan.Connected)
|
||||
sort.Slice(copyPlan.NoShows, func(i, j int) bool { return copyPlan.NoShows[i].PlayerID < copyPlan.NoShows[j].PlayerID })
|
||||
b, err := json.Marshal(struct {
|
||||
MatchID string
|
||||
Plan domain.InitialConnectPlan
|
||||
}{matchID, copyPlan})
|
||||
if err != nil {
|
||||
return [32]byte{}, err
|
||||
}
|
||||
return sha256.Sum256(b), nil
|
||||
}
|
||||
|
||||
func loadInitialConnectParticipants(ctx context.Context, tx *sql.Tx, matchID string) ([]initialConnectParticipant, error) {
|
||||
rows, err := tx.QueryContext(ctx, initialConnectParticipantsSQL, matchID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var result []initialConnectParticipant
|
||||
for rows.Next() {
|
||||
var p initialConnectParticipant
|
||||
if err := rows.Scan(&p.PlayerID, &p.TicketID, &p.Team, &p.ConnectedAt, &p.Active); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, p)
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
func validateInitialConnectPlan(plan domain.InitialConnectPlan, participants []initialConnectParticipant, playlist domain.Playlist) error {
|
||||
if len(participants) == 0 || (plan.Action == domain.InitialConnectStartWithBot && playlist != domain.Casual) || (plan.Action == domain.InitialConnectCancel && plan.MatchState != domain.Cancelled) {
|
||||
return fmt.Errorf("invalid initial-connect plan")
|
||||
}
|
||||
known, connected, missing := map[string]bool{}, map[string]bool{}, map[string]bool{}
|
||||
for _, p := range participants {
|
||||
if p.PlayerID == "" || !p.Active || known[p.PlayerID] {
|
||||
return fmt.Errorf("invalid stored participant roster")
|
||||
}
|
||||
known[p.PlayerID] = true
|
||||
if p.ConnectedAt.Valid {
|
||||
connected[p.PlayerID] = true
|
||||
}
|
||||
}
|
||||
for _, id := range plan.Connected {
|
||||
if !known[id] || !connected[id] || missing[id] {
|
||||
return fmt.Errorf("invalid connected participant")
|
||||
}
|
||||
missing[id] = true
|
||||
}
|
||||
for _, noShow := range plan.NoShows {
|
||||
if !known[noShow.PlayerID] || connected[noShow.PlayerID] || missing[noShow.PlayerID] || noShow.Cooldown <= 0 || noShow.AbandonedAt.IsZero() {
|
||||
return fmt.Errorf("invalid no-show participant")
|
||||
}
|
||||
missing[noShow.PlayerID] = true
|
||||
}
|
||||
if len(missing) != len(known) {
|
||||
return fmt.Errorf("initial-connect plan does not cover roster")
|
||||
}
|
||||
if plan.Action == domain.InitialConnectStartWithBot {
|
||||
if len(plan.CasualLineup) != 6 {
|
||||
return fmt.Errorf("casual bot lineup must contain six players")
|
||||
}
|
||||
lineupSlots := make(map[int]bool, 6)
|
||||
lineupPlayers := make(map[string]bool, 6)
|
||||
for _, slot := range plan.CasualLineup {
|
||||
if slot.Slot < 0 || slot.Slot > 5 || slot.Team != slot.Slot%2 || lineupSlots[slot.Slot] || slot.PlayerID == "" || lineupPlayers[slot.PlayerID] {
|
||||
return fmt.Errorf("invalid casual bot lineup")
|
||||
}
|
||||
lineupSlots[slot.Slot] = true
|
||||
lineupPlayers[slot.PlayerID] = true
|
||||
if slot.IsBot {
|
||||
continue
|
||||
}
|
||||
if !connected[slot.PlayerID] {
|
||||
return fmt.Errorf("lineup contains non-connected human")
|
||||
}
|
||||
}
|
||||
for id := range connected {
|
||||
if !lineupPlayers[id] {
|
||||
return fmt.Errorf("lineup omits connected human")
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user