CosmicClash/server/store/allocation_match_sql.go

package store

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
)

const AllocationClaimLease = time.Minute

type PendingAllocation struct {
	Request domain.AllocationRequest
}

const ClaimAllocatingMatchSQL = `WITH candidate AS (
    SELECT match_id FROM matches
    WHERE state = 'ALLOCATING' AND server_id IS NULL
      AND (allocation_id IS NULL OR allocation_claimed_at <= $1)
    ORDER BY created_at, match_id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE matches m
SET allocation_id = 'allocation-' || candidate.match_id, allocation_claimed_at = $2
FROM candidate
WHERE m.match_id = candidate.match_id
RETURNING m.match_id, m.region, m.protocol_version, m.allocation_id`

const AllocatingMatchBuildSQL = `SELECT client_build
FROM queue_tickets q
JOIN match_participants mp ON mp.ticket_id = q.ticket_id AND mp.player_id = q.player_id
WHERE mp.match_id = $1
ORDER BY q.client_build`

const BindAllocatedMatchParticipantsSQL = `WITH bound AS (
    UPDATE matches
    SET server_id = $3
    WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL
      AND EXISTS (
          SELECT 1 FROM allocations
          WHERE allocation_id = $2 AND match_id = $1 AND server_id = $3 AND state = 'ALLOCATED'
      )
    RETURNING match_id
), participants AS (
    SELECT mp.ticket_id, mp.player_id
    FROM match_participants mp
    JOIN bound ON bound.match_id = mp.match_id
), advanced AS (
    UPDATE queue_tickets q
    SET state = 'ALLOCATING', revision = revision + 1
    FROM participants p
    WHERE q.ticket_id = p.ticket_id AND q.player_id = p.player_id AND q.state = 'ACCEPTED'
    RETURNING q.ticket_id
)
SELECT (SELECT count(*) FROM participants), (SELECT count(*) FROM advanced)`

const ReleaseAllocatedMatchClaimSQL = `UPDATE matches
SET allocation_id = NULL, allocation_claimed_at = NULL
WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL`

// FindProviderAllocation verifies whether a recovered lease has already
// crossed the durable provider boundary. A worker can then bind it without
// issuing a second external allocation request after a crash.
func FindProviderAllocation(ctx context.Context, db *sql.DB, request domain.AllocationRequest) (domain.Allocation, bool, error) {
	if db == nil || request.AllocationID == "" || request.MatchID == "" || request.Region == "" || request.Build == "" || request.Protocol <= 0 || (request.Transport != "enet" && request.Transport != "steam_sdr") {
		return domain.Allocation{}, false, fmt.Errorf("invalid provider allocation lookup")
	}
	var allocation domain.Allocation
	var digest []byte
	err := db.QueryRowContext(ctx, SelectAllocationSQL, request.AllocationID).Scan(&allocation.AllocationID, &allocation.MatchID, &allocation.ServerID, &allocation.Region, &allocation.Build, &allocation.Protocol, &allocation.Transport, &allocation.AllocatedAt, &digest)
	if err == sql.ErrNoRows {
		return domain.Allocation{}, false, nil
	}
	if err != nil {
		return domain.Allocation{}, false, err
	}
	want := allocationRequestDigest(request)
	if !bytes.Equal(digest, want[:]) || allocation.MatchID != request.MatchID || allocation.Region != request.Region || allocation.Build != request.Build || allocation.Protocol != request.Protocol || allocation.Transport != request.Transport {
		return domain.Allocation{}, false, domain.ErrConflict
	}
	allocation.State = domain.ServerAllocated
	return allocation, true, nil
}

// ClaimAllocatingMatch returns one durable provider work item. The fixed
// allocation ID is retained across a lease recovery, allowing every later
// reconciliation step to reject a different server for the same match.
func ClaimAllocatingMatch(ctx context.Context, db *sql.DB, transport string, now time.Time) (PendingAllocation, bool, error) {
	if db == nil || (transport != "enet" && transport != "steam_sdr") || now.IsZero() {
		return PendingAllocation{}, false, fmt.Errorf("invalid allocation claim arguments")
	}
	var item PendingAllocation
	found := false
	err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
		var matchID, region string
		var protocol int
		var claimedID string
		err := tx.QueryRowContext(ctx, ClaimAllocatingMatchSQL, now.Add(-AllocationClaimLease), now).Scan(&matchID, &region, &protocol, &claimedID)
		if err == sql.ErrNoRows {
			return nil
		}
		if err != nil {
			return err
		}
		rows, err := tx.QueryContext(ctx, AllocatingMatchBuildSQL, matchID)
		if err != nil {
			return err
		}
		defer rows.Close()
		build := ""
		for rows.Next() {
			var candidate string
			if err := rows.Scan(&candidate); err != nil {
				return err
			}
			if build == "" {
				build = candidate
			} else if build != candidate {
				return fmt.Errorf("allocating match has mixed client builds")
			}
		}
		if err := rows.Err(); err != nil {
			return err
		}
		if build == "" {
			return fmt.Errorf("allocating match has no participants")
		}
		item.Request = domain.AllocationRequest{AllocationID: claimedID, MatchID: matchID, Region: region, Build: build, Protocol: protocol, Transport: transport}
		found = true
		return nil
	})
	return item, found, err
}

func BindAllocatedMatch(ctx context.Context, db *sql.DB, allocation domain.Allocation) error {
	if db == nil || allocation.MatchID == "" || allocation.AllocationID == "" || allocation.ServerID == "" || allocation.State != domain.ServerAllocated {
		return fmt.Errorf("invalid allocated match binding")
	}
	return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
		var participants, advanced int
		if err := tx.QueryRowContext(ctx, BindAllocatedMatchParticipantsSQL, allocation.MatchID, allocation.AllocationID, allocation.ServerID).Scan(&participants, &advanced); err != nil {
			return err
		}
		if participants == 0 || participants != advanced {
			return domain.ErrConflict
		}
		return nil
	})
}

func ReleaseAllocatedMatchClaim(ctx context.Context, db *sql.DB, matchID, allocationID string) error {
	if db == nil || matchID == "" || allocationID == "" {
		return fmt.Errorf("invalid allocated match claim release")
	}
	result, err := db.ExecContext(ctx, ReleaseAllocatedMatchClaimSQL, matchID, allocationID)
	if err != nil {
		return err
	}
	changed, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if changed != 1 {
		return domain.ErrConflict
	}
	return nil
}