CosmicClash/server/store/queue_sql.go

package store

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"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 IN ('QUEUED', 'PROPOSED')
RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision, predicted_rtt`
	QueueMutationFailureSQL = `SELECT player_id, state, revision, expires_at
FROM queue_tickets
WHERE ticket_id = $1
FOR UPDATE`
	QueueCooldownSelectSQL = `SELECT ends_at
FROM penalties
WHERE player_id = $1 AND playlist = $2
  AND kind IN ('PROPOSAL_DECLINED', 'PROPOSAL_TIMEOUT', 'INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED')
  AND ends_at > $3
ORDER BY ends_at DESC
LIMIT 1`
)

// QueueCandidateProjectionSQL joins the authoritative rating. Without it every
// PostgreSQL-sourced candidate carried Go's zero value, and since rating
// tolerance, selection scoring and team partitioning all read that field,
// ranked matchmaking treated every player as identically rated. A player with
// no ratings row yet is a genuinely new profile and starts at the Glicko
// initial rating, matching domain.GlickoInitialRating and the column default.
// The rating is never taken from the client.
const QueueCandidateProjectionSQL = `SELECT q.ticket_id, q.player_id, q.playlist, q.client_build,
    q.protocol_version, q.enqueued_at, q.predicted_rtt, COALESCE(r.rating, $4)
FROM queue_tickets q
LEFT JOIN ratings r ON r.player_id = q.player_id
WHERE q.state = 'QUEUED' AND q.playlist = $1 AND q.expires_at > $2
ORDER BY q.enqueued_at, q.ticket_id
LIMIT $3`

// QueueTicketRatingSQL resolves a player's authoritative rating, falling back
// to the new-profile default when they have no ratings row yet.
const QueueTicketRatingSQL = `SELECT COALESCE((SELECT rating FROM ratings WHERE player_id = $1), $2)`

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, domain.GlickoInitialRating)
	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, &candidate.Rating); 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 {
		// The Redis projection is seeded from this candidate, so it must carry
		// the same authoritative rating the durable candidate query joins.
		// Reading it inside the transaction keeps both projections agreeing on
		// one value rather than one of them defaulting to zero.
		var rating float64
		if err := tx.QueryRowContext(ctx, QueueTicketRatingSQL, playerID, domain.GlickoInitialRating).Scan(&rating); err != nil {
			return err
		}
		candidate := domain.Candidate{TicketID: ticketID, PlayerID: playerID, Playlist: spec.Playlist, ClientBuild: spec.ClientBuild, ProtocolVersion: spec.ProtocolVersion, EnqueuedAt: now, Rating: rating}
		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[:]) {
				// Must wrap ErrConflict: writeDomainError maps unrecognised
				// errors to 422, but the contract
				// (state-transitions.json "same_key_different_payload") and
				// openapi.json both require 409 for reusing a key with a
				// different payload.
				return fmt.Errorf("%w: queue create idempotency conflict", domain.ErrConflict)
			}
			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))
		}
		// A nil map marshals to JSON `null`, a JSONB scalar -- not an empty
		// object. jsonb_set then fails with "cannot set path in scalar", so
		// the first probe for this player could never be recorded even once
		// the probe endpoint was wired. Persist an object from the start.
		if candidate.PredictedRTT == nil {
			candidate.PredictedRTT = map[string]float64{}
			ticket.Candidate.PredictedRTT = candidate.PredictedRTT
		}
		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
-- COALESCE only guards SQL NULL. Rows written before the insert fix hold a
-- JSONB scalar null, which jsonb_set rejects outright, so normalise anything
-- that is not an object before setting the region key.
SET predicted_rtt = jsonb_set(
    CASE WHEN jsonb_typeof(COALESCE(predicted_rtt, '{}'::jsonb)) = 'object'
         THEN predicted_rtt ELSE '{}'::jsonb END,
    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("%w: queue mutation idempotency conflict", domain.ErrConflict)
			}
			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 {
			if !errors.Is(err, sql.ErrNoRows) {
				return fmt.Errorf("queue mutation rejected: %w", err)
			}
			return classifyQueueMutationFailure(ctx, tx, playerID, ticketID, expectedRevision, now)
		}
		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 classifyQueueMutationFailure(ctx context.Context, tx *sql.Tx, playerID, ticketID string, expectedRevision uint64, now time.Time) error {
	var owner, state string
	var revision uint64
	var expiresAt time.Time
	err := tx.QueryRowContext(ctx, QueueMutationFailureSQL, ticketID).Scan(&owner, &state, &revision, &expiresAt)
	if errors.Is(err, sql.ErrNoRows) {
		return domain.ErrTicketNotFound
	}
	if err != nil {
		return err
	}
	if owner != playerID {
		return domain.ErrNotTicketOwner
	}
	if (state == string(domain.Queued) || state == string(domain.Proposed)) && !now.Before(expiresAt) {
		return domain.ErrTicketExpired
	}
	if revision != expectedRevision {
		return domain.ErrStaleRevision
	}
	return fmt.Errorf("%w: %s in %s", domain.ErrConflict, "queue mutation", state)
}

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

// QueueCandidateByPlayerSQL mirrors QueueCandidateProjectionSQL for a single
// player, so a probe can repair that player's transient index entry without
// re-reading the whole queue.
const QueueCandidateByPlayerSQL = `SELECT q.ticket_id, q.player_id, q.playlist, q.client_build,
    q.protocol_version, q.enqueued_at, q.predicted_rtt, COALESCE(r.rating, $3)
FROM queue_tickets q
LEFT JOIN ratings r ON r.player_id = q.player_id
WHERE q.player_id = $1 AND q.state = 'QUEUED' AND q.expires_at > $2`

// FindQueuedCandidateByPlayer returns the player's live queue candidate, if
// any. The second result reports whether the player is currently queued; a
// player who is not queued is not an error.
func FindQueuedCandidateByPlayer(ctx context.Context, db *sql.DB, playerID string, now time.Time) (domain.Candidate, bool, error) {
	if db == nil || playerID == "" || now.IsZero() {
		return domain.Candidate{}, false, fmt.Errorf("invalid queued candidate lookup")
	}
	var candidate domain.Candidate
	var playlist string
	var predictedRTT []byte
	err := db.QueryRowContext(ctx, QueueCandidateByPlayerSQL, playerID, now, domain.GlickoInitialRating).
		Scan(&candidate.TicketID, &candidate.PlayerID, &playlist, &candidate.ClientBuild, &candidate.ProtocolVersion, &candidate.EnqueuedAt, &predictedRTT, &candidate.Rating)
	if err == sql.ErrNoRows {
		return domain.Candidate{}, false, nil
	}
	if err != nil {
		return domain.Candidate{}, false, err
	}
	if err := json.Unmarshal(predictedRTT, &candidate.PredictedRTT); err != nil {
		return domain.Candidate{}, false, fmt.Errorf("decode candidate RTT: %w", err)
	}
	candidate.Playlist = domain.Playlist(playlist)
	return candidate, true, nil
}