CosmicClash/server/matcher/worker.go

// Package matcher contains the provider-neutral orchestration around the
// durable proposal claim transaction.
package matcher

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// These three are the only RunOnce failures Run treats as fatal to the whole
// worker: they're static misconfiguration, true on every future pass just as
// much as this one, so retrying cannot help. Every other RunOnce error --
// a source read hiccup, no common region among the current candidate pool,
// a losing race against another matcher replica, an incomplete batch -- is a
// single pass's worth of "no match formed this time," a routine and
// expected steady state that must not take matching down for every other
// player still waiting behind it.
var (
	ErrWorkerNotConfigured = errors.New("matcher worker is not configured")
	ErrUnsupportedPlaylist = errors.New("unsupported matcher playlist")
	ErrInvalidMatcherSize  = errors.New("invalid matcher size")
)

type CandidateSource func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error)

type ProposalCreator interface {
	CreateProposal(context.Context, domain.Proposal, map[string]string, time.Time) error
}

type ProposalCreatorFunc func(context.Context, domain.Proposal, map[string]string, time.Time) error

func (f ProposalCreatorFunc) CreateProposal(ctx context.Context, proposal domain.Proposal, ticketIDs map[string]string, now time.Time) error {
	return f(ctx, proposal, ticketIDs, now)
}

type PrepareFunc func(string, domain.Playlist, domain.MatchFormation, time.Time) (domain.PreparedProposal, error)

type Worker struct {
	Source   CandidateSource
	Creator  ProposalCreator
	Playlist domain.Playlist
	Size     int
	Now      func() time.Time
	NextID   func() string
	Prepare  PrepareFunc
	OnError  func(error)
}

// Run polls until cancellation. A failed attempt is returned so a supervisor
// can restart the role rather than silently dropping durable claim failures.
func (w Worker) Run(ctx context.Context, interval time.Duration) error {
	if interval <= 0 {
		return fmt.Errorf("matcher interval must be positive")
	}
	for {
		if _, err := w.RunOnce(ctx); err != nil {
			if w.OnError != nil {
				w.OnError(err)
			}
			if errors.Is(err, ErrWorkerNotConfigured) || errors.Is(err, ErrUnsupportedPlaylist) || errors.Is(err, ErrInvalidMatcherSize) {
				return err
			}
			// Not fatal -- fall through and retry next interval.
		}
		timer := time.NewTimer(interval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil
		case <-timer.C:
		}
	}
}

// RunOnce performs one bounded matchmaking attempt. The source may be Redis
// backed, but the creator must be the durable transaction that claims tickets;
// a stale cache therefore fails safely and can be retried on the next pass.
func (w Worker) RunOnce(ctx context.Context) (bool, error) {
	if w.Source == nil || w.Creator == nil || w.Now == nil || w.NextID == nil || w.Prepare == nil {
		return false, ErrWorkerNotConfigured
	}
	if w.Playlist != domain.Casual && w.Playlist != domain.Ranked {
		return false, ErrUnsupportedPlaylist
	}
	if w.Size < 2 || w.Size > 6 {
		return false, ErrInvalidMatcherSize
	}
	now := w.Now()
	candidates, err := w.Source(ctx, now, w.Playlist, w.Size)
	if err != nil {
		return false, err
	}
	if len(candidates) < w.Size {
		return false, nil
	}
	queue := domain.NewQueue()
	for _, candidate := range candidates {
		if candidate.Playlist != w.Playlist {
			return false, fmt.Errorf("candidate playlist does not match worker")
		}
		if _, err := queue.Create(candidate.PlayerID, candidate.TicketID, "matcher-"+candidate.TicketID, candidate, now); err != nil {
			return false, err
		}
	}
	formation, err := domain.FormFromQueue(queue, w.Size, now)
	if err != nil {
		return false, err
	}
	prepared, err := w.Prepare(w.NextID(), w.Playlist, formation, now)
	if err != nil {
		return false, err
	}
	ticketIDs := make(map[string]string, len(prepared.Proposal.Participants))
	for _, participant := range prepared.Proposal.Participants {
		for _, candidate := range formation.Selection.Players {
			if candidate.PlayerID == participant.PlayerID {
				ticketIDs[participant.PlayerID] = candidate.TicketID
				break
			}
		}
	}
	if len(ticketIDs) != len(prepared.Proposal.Participants) {
		return false, fmt.Errorf("proposal participant is not in formed selection")
	}
	if err := w.Creator.CreateProposal(ctx, prepared.Proposal, ticketIDs, now); err != nil {
		return false, err
	}
	return true, nil
}