hockey_new/agent/dispatch.go

package main

import (
	"context"
	"encoding/json"
	"math"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	dispatchControlCapacity = 64
	dispatchBulkCapacity    = 2
	dispatchCommandLimit    = 500
	dispatchBulkWorkers     = 5
	dispatchHTTPWorkers     = 6
	dispatchMaxDeadline     = 59 * time.Second
)

// Only idempotent field setters belong in the background Mapping lane.
var mappingBulkFunctions = map[string]bool{
	"SetText": true, "SetImage": true, "SetColor": true, "SetTextColour": true,
	"SetTextVisibleOn": true, "SetTextVisibleOff": true,
	"SetImageVisibleOn": true, "SetImageVisibleOff": true,
}

type dispatchResult struct {
	OK     bool   `json:"ok"`
	Reason string `json:"reason"`
}

type dispatchJob struct {
	ctx           context.Context
	cancel        context.CancelFunc
	epoch         uint64
	assignmentID  string
	matchID       string
	vmixURL       string
	requestID     string
	batch         bool
	bulk          bool
	commands      []map[string]any
	invalidReason string
	timerRevision uint64
}

// One dispatcher belongs to one authenticated WebSocket connection. Its two
// coordinators and bounded per-job chain workers never grow with packet count.
type commandDispatcher struct {
	agent            *Agent
	ws               *wsConn
	ctx              context.Context
	cancel           context.CancelFunc
	mu               sync.Mutex
	assignment       context.Context
	cancelAssignment context.CancelFunc
	epoch            uint64
	assignmentID     string
	matchID          string
	controls         chan *dispatchJob
	bulk             chan *dispatchJob
	permits          chan struct{}
	controlPending   int
	changed          chan struct{}
	workers          sync.WaitGroup
	timerRevision    uint64
	targetRevisions  map[string]uint64
	targetLocks      map[string]chan struct{}
}

func newCommandDispatcher(a *Agent, ws *wsConn) *commandDispatcher {
	ctx, cancel := context.WithCancel(context.Background())
	assignment, cancelAssignment := context.WithCancel(ctx)
	a.mu.RLock()
	assignmentID, matchID := a.state.AssignmentID, a.state.MatchID
	a.mu.RUnlock()
	d := &commandDispatcher{agent: a, ws: ws, ctx: ctx, cancel: cancel,
		assignment: assignment, cancelAssignment: cancelAssignment,
		assignmentID: assignmentID, matchID: matchID,
		controls: make(chan *dispatchJob, dispatchControlCapacity),
		bulk:     make(chan *dispatchJob, dispatchBulkCapacity),
		permits:  make(chan struct{}, dispatchHTTPWorkers), changed: make(chan struct{}),
		targetRevisions: map[string]uint64{}, targetLocks: map[string]chan struct{}{}}
	d.workers.Add(2)
	go d.runLane(d.controls, false)
	go d.runLane(d.bulk, true)
	return d
}

func (d *commandDispatcher) close() {
	d.cancel()
	d.workers.Wait()
}

func (d *commandDispatcher) notifyLocked() {
	close(d.changed)
	d.changed = make(chan struct{})
}

func (d *commandDispatcher) replaceAssignment(assignmentID, matchID string) {
	d.mu.Lock()
	if d.assignmentID != assignmentID || d.matchID != matchID {
		d.cancelAssignment()
		d.assignment, d.cancelAssignment = context.WithCancel(d.ctx)
		d.epoch++
		d.assignmentID, d.matchID = assignmentID, matchID
		d.notifyLocked()
	}
	d.mu.Unlock()
}

// Non-command messages stay immediate. In particular, assignment changes cancel
// running HTTP requests before updating the Agent state or accepting more work.
func (d *commandDispatcher) handle(msg map[string]any) error {
	typ := stringValue(msg["type"])
	switch typ {
	case "vmix.command", "vmix.batch":
		return d.enqueue(msg)
	case "match.assign":
		d.replaceAssignment(stringValue(msg["assignment_id"]), stringValue(msg["game_id"]))
	case "hello.ok":
		if device, ok := msg["device"].(map[string]any); ok {
			d.replaceAssignment(stringValue(device["assignment_id"]), stringValue(device["current_match_id"]))
		}
	case "pairing.revoked", "device.deactivated":
		d.replaceAssignment("", "")
	}
	return d.agent.handleMessage(d.ws, msg)
}

func dispatchDeadline(raw any) (time.Duration, bool) {
	if raw == nil {
		return 0, false
	}
	var value float64
	switch n := raw.(type) {
	case float64:
		value = n
	case int:
		value = float64(n)
	case int64:
		value = float64(n)
	case json.Number:
		parsed, err := strconv.ParseFloat(string(n), 64)
		if err != nil {
			return 0, false
		}
		value = parsed
	default:
		return 0, false
	}
	if math.IsNaN(value) || math.IsInf(value, 0) || value <= 0 || value != math.Trunc(value) {
		return 0, false
	}
	if value > float64(dispatchMaxDeadline/time.Millisecond) {
		return dispatchMaxDeadline, true
	}
	return time.Duration(value) * time.Millisecond, true
}

func (d *commandDispatcher) enqueue(msg map[string]any) error {
	batch := stringValue(msg["type"]) == "vmix.batch"
	commands := []map[string]any{}
	if batch {
		raw, ok := msg["commands"].([]any)
		if !ok {
			return d.reject(msg, nil, "invalid_commands")
		}
		if len(raw) > dispatchCommandLimit {
			return d.reject(msg, nil, "too_many_commands")
		}
		for _, value := range raw {
			command, _ := value.(map[string]any)
			commands = append(commands, command)
		}
	} else {
		command, _ := msg["command"].(map[string]any)
		commands = append(commands, command)
	}
	bulk := false
	if raw, present := msg["mapping_bulk"]; present {
		flag, ok := raw.(bool)
		if !ok || (flag && !batch) {
			return d.reject(msg, commands, "invalid_mapping_bulk")
		}
		bulk = flag
	}
	deadline := dispatchMaxDeadline
	if bulk {
		assignmentID, assignmentOK := msg["assignment_id"].(string)
		matchID, matchOK := msg["match_id"].(string)
		if !assignmentOK || !matchOK || strings.TrimSpace(assignmentID) == "" || strings.TrimSpace(matchID) == "" {
			return d.reject(msg, commands, "missing_mapping_scope")
		}
		if raw, present := msg["mapping_deadline_ms"]; present {
			var ok bool
			deadline, ok = dispatchDeadline(raw)
			if !ok {
				return d.reject(msg, commands, "invalid_mapping_deadline")
			}
		}
	}
	d.mu.Lock()
	ctx, cancel := context.WithTimeout(d.assignment, deadline)
	job := &dispatchJob{ctx: ctx, cancel: cancel, epoch: d.epoch,
		assignmentID: d.assignmentID, matchID: d.matchID,
		requestID: stringValue(msg["request_id"]), batch: batch, bulk: bulk, commands: commands,
		timerRevision: d.timerRevision}
	if incoming := stringValue(msg["assignment_id"]); incoming != "" {
		job.assignmentID = incoming
		if incoming != d.assignmentID {
			job.invalidReason = "assignment_mismatch"
		}
	}
	if incoming := stringValue(msg["match_id"]); incoming != "" {
		job.matchID = incoming
		if incoming != d.matchID {
			job.invalidReason = "match_mismatch"
		}
	}
	d.agent.mu.RLock()
	job.vmixURL = d.agent.cfg.VmixURL
	currentScope := d.agent.currentWS == d.ws && d.agent.state.AssignmentID == job.assignmentID && d.agent.state.MatchID == job.matchID && job.ctx.Err() == nil && d.ctx.Err() == nil
	d.agent.mu.RUnlock()
	if !bulk {
		d.controlPending++
		d.notifyLocked()
	}
	lane := d.controls
	if bulk {
		lane = d.bulk
	}
	select {
	case lane <- job:
		// Fence only accepted, authorized timer work. Rejected/overflow packets
		// cannot suppress an otherwise valid Mapping field.
		if !bulk && job.invalidReason == "" && currentScope {
			for _, command := range commands {
				if dispatchTimerFunction(dispatchFunction(command)) {
					d.timerRevision++
					for _, target := range d.commandTargetKeys(command) {
						d.targetRevisions[target] = d.timerRevision
					}
				}
			}
		}
		d.mu.Unlock()
		return nil
	default:
		if !bulk {
			d.controlPending--
			d.notifyLocked()
		}
		d.mu.Unlock()
		cancel()
		reason := "control_queue_full"
		if bulk {
			reason = "mapping_queue_full"
		}
		return d.reject(msg, commands, reason)
	}
}

func (d *commandDispatcher) reject(msg map[string]any, commands []map[string]any, reason string) error {
	rows := make([]dispatchResult, len(commands))
	for i := range rows {
		rows[i] = dispatchResult{Reason: reason}
	}
	batch := stringValue(msg["type"]) == "vmix.batch"
	payload := map[string]any{"type": "command.ack", "request_id": stringValue(msg["request_id"]),
		"assignment_id": stringValue(msg["assignment_id"]), "match_id": stringValue(msg["match_id"]),
		"ok": false, "reason": reason}
	if batch {
		payload["type"] = "command.batch.ack"
		payload["results"] = rows
		payload["applied"] = 0
		payload["failed"] = len(rows)
	}
	return d.write(payload)
}

func (d *commandDispatcher) runLane(lane <-chan *dispatchJob, bulk bool) {
	defer d.workers.Done()
	for {
		select {
		case <-d.ctx.Done():
			return
		case job := <-lane:
			d.runJob(job)
			job.cancel()
			if !bulk {
				d.mu.Lock()
				d.controlPending--
				d.notifyLocked()
				d.mu.Unlock()
			}
		}
	}
}

func (d *commandDispatcher) reason(job *dispatchJob) string {
	if d.ctx.Err() != nil {
		return "connection_closed"
	}
	d.mu.Lock()
	same := job.epoch == d.epoch && job.assignmentID == d.assignmentID && job.matchID == d.matchID
	d.mu.Unlock()
	if job.invalidReason != "" {
		return job.invalidReason
	}
	if !same {
		return "assignment_changed"
	}
	if job.ctx.Err() == context.DeadlineExceeded {
		return "mapping_deadline_exceeded"
	}
	if job.ctx.Err() != nil {
		return "assignment_changed"
	}
	d.agent.mu.RLock()
	same = d.agent.currentWS == d.ws && d.agent.state.AssignmentID == job.assignmentID && d.agent.state.MatchID == job.matchID && d.agent.cfg.VmixURL == job.vmixURL
	d.agent.mu.RUnlock()
	if !same {
		return "stale_command_scope"
	}
	return ""
}

func (d *commandDispatcher) acquire(job *dispatchJob) bool {
	if !job.bulk {
		select {
		case d.permits <- struct{}{}:
			return true
		case <-job.ctx.Done():
			return false
		}
	}
	for {
		d.mu.Lock()
		changed := d.changed
		if d.controlPending == 0 {
			select {
			case d.permits <- struct{}{}:
				d.mu.Unlock()
				return true
			default:
			}
		}
		d.mu.Unlock()
		select {
		case <-changed:
		case <-job.ctx.Done():
			return false
		}
	}
}

func (d *commandDispatcher) release() {
	<-d.permits
	d.mu.Lock()
	d.notifyLocked()
	d.mu.Unlock()
}

func dispatchFunction(command map[string]any) string {
	function := stringValue(command["Function"])
	if function == "" {
		function = stringValue(command["function"])
	}
	return function
}

func dispatchTimerFunction(function string) bool {
	switch function {
	case "SetCountdown", "StartCountdown", "PauseCountdown", "StopCountdown":
		return true
	}
	return false
}

// Include every matching inventory identity. A duplicate title conservatively
// joins its possible Inputs instead of allowing a title/number alias bypass.
func (d *commandDispatcher) commandTargetKeys(command map[string]any) []string {
	input := strings.TrimSpace(stringValue(command["Input"]))
	field := strings.ToLower(strings.TrimSpace(stringValue(command["SelectedName"])))
	aliases := map[string]bool{strings.ToLower(input): true}
	d.agent.mu.RLock()
	for _, item := range d.agent.inventory.Inputs {
		if strings.EqualFold(input, item.Key) || strings.EqualFold(input, item.Number) || strings.EqualFold(input, item.Title) {
			for _, alias := range []string{item.Key, item.Number, item.Title} {
				if alias = strings.TrimSpace(alias); alias != "" {
					aliases[strings.ToLower(alias)] = true
				}
			}
		}
	}
	d.agent.mu.RUnlock()
	keys := make([]string, 0, len(aliases))
	for alias := range aliases {
		keys = append(keys, alias+"\x00"+field)
	}
	sort.Strings(keys)
	return keys
}

func (d *commandDispatcher) lockTargets(job *dispatchJob, keys []string) (func(), bool) {
	d.mu.Lock()
	locks := make([]chan struct{}, len(keys))
	for i, key := range keys {
		lock := d.targetLocks[key]
		if lock == nil {
			lock = make(chan struct{}, 1)
			d.targetLocks[key] = lock
		}
		locks[i] = lock
	}
	d.mu.Unlock()
	held := 0
	release := func() {
		for i := held - 1; i >= 0; i-- {
			<-locks[i]
		}
	}
	for _, lock := range locks {
		select {
		case lock <- struct{}{}:
			held++
		case <-job.ctx.Done():
			release()
			return nil, false
		}
	}
	return release, true
}

func (d *commandDispatcher) execute(job *dispatchJob, command map[string]any) dispatchResult {
	if reason := d.reason(job); reason != "" {
		return dispatchResult{Reason: reason}
	}
	function := dispatchFunction(command)
	if function == "" {
		return dispatchResult{Reason: "missing_function"}
	}
	if job.bulk && !mappingBulkFunctions[function] {
		return dispatchResult{Reason: "unsafe_mapping_function"}
	}
	if job.bulk && (strings.TrimSpace(stringValue(command["Input"])) == "" || strings.TrimSpace(stringValue(command["SelectedName"])) == "") {
		return dispatchResult{Reason: "missing_mapping_target"}
	}
	if !d.acquire(job) {
		return dispatchResult{Reason: d.reason(job)}
	}
	defer d.release()
	targets := d.commandTargetKeys(command)
	unlockTargets, locked := d.lockTargets(job, targets)
	if !locked {
		return dispatchResult{Reason: d.reason(job)}
	}
	defer unlockTargets()
	if reason := d.reason(job); reason != "" {
		return dispatchResult{Reason: reason}
	}
	if job.bulk {
		d.mu.Lock()
		protected := false
		for _, target := range targets {
			if d.targetRevisions[target] > job.timerRevision {
				protected = true
				break
			}
		}
		d.mu.Unlock()
		if protected {
			return dispatchResult{Reason: "native_countdown_protected"}
		}
	}
	params := url.Values{}
	params.Set("Function", function)
	for key, value := range command {
		if strings.EqualFold(key, "function") || value == nil {
			continue
		}
		text := stringValue(value)
		if strings.EqualFold(key, "Value") || text != "" {
			params.Set(key, text)
		}
	}
	requestCtx, cancel := context.WithTimeout(job.ctx, 3*time.Second)
	status := probeVmixContext(requestCtx, job.vmixURL, params)
	cancel()
	if reason := d.reason(job); reason != "" {
		return dispatchResult{Reason: reason}
	}
	// State updates share the authority lock ordering with scope checks.
	d.mu.Lock()
	d.agent.mu.Lock()
	current := job.epoch == d.epoch && job.ctx.Err() == nil && d.ctx.Err() == nil && d.agent.currentWS == d.ws &&
		d.agent.state.AssignmentID == job.assignmentID && d.agent.state.MatchID == job.matchID && d.agent.cfg.VmixURL == job.vmixURL
	if current {
		// A bad field is a command failure, not a failed base-API health probe.
		if status.Connected {
			d.agent.state.Vmix = status
		}
		d.agent.state.LastCommand = function
		d.agent.state.LastCommandOK = status.Connected
		d.agent.state.LastCommandAt = time.Now()
		d.agent.state.LastCommandDetail = status.Error
	}
	d.agent.mu.Unlock()
	d.mu.Unlock()
	if !current {
		return dispatchResult{Reason: "stale_command_scope"}
	}
	return dispatchResult{OK: status.Connected, Reason: status.Error}
}

func (d *commandDispatcher) runJob(job *dispatchJob) {
	results := make([]dispatchResult, len(job.commands))
	for i := range results {
		results[i].Reason = "mapping_not_attempted"
	}
	// Chains preserve caller order for one physical target; different targets
	// may use the bounded HTTP pool concurrently, including legacy batches.
	chains := [][]int{}
	parents := make([]int, len(job.commands))
	var root func(int) int
	root = func(index int) int {
		if parents[index] != index {
			parents[index] = root(parents[index])
		}
		return parents[index]
	}
	targets := map[string]int{}
	for index, command := range job.commands {
		parents[index] = index
		for _, target := range d.commandTargetKeys(command) {
			if previous, exists := targets[target]; exists {
				parents[root(index)] = root(previous)
			} else {
				targets[target] = index
			}
		}
	}
	groups := map[int]int{}
	for index := range job.commands {
		parent := root(index)
		chain, exists := groups[parent]
		if !exists {
			chain = len(chains)
			groups[parent] = chain
			chains = append(chains, []int{})
		}
		chains[chain] = append(chains[chain], index)
	}
	workers := dispatchHTTPWorkers
	if job.bulk {
		workers = dispatchBulkWorkers
	}
	if len(chains) < workers {
		workers = len(chains)
	}
	work := make(chan []int)
	var group sync.WaitGroup
	group.Add(workers)
	for i := 0; i < workers; i++ {
		go func() {
			defer group.Done()
			for chain := range work {
				for _, index := range chain {
					results[index] = d.execute(job, job.commands[index])
				}
			}
		}()
	}
	for _, chain := range chains {
		select {
		case work <- chain:
		case <-job.ctx.Done():
			// Workers still drain bounded chains with cancelled scope, producing
			// explicit false rows; no command is silently treated as successful.
			work <- chain
		}
	}
	close(work)
	group.Wait()
	if reason := d.reason(job); reason != "" {
		for i := range results {
			if reason != "mapping_deadline_exceeded" || results[i].Reason == "mapping_not_attempted" {
				results[i] = dispatchResult{Reason: reason}
			}
		}
	}
	applied := 0
	for _, row := range results {
		if row.OK {
			applied++
		}
	}
	payload := map[string]any{"type": "command.ack", "request_id": job.requestID,
		"assignment_id": job.assignmentID, "match_id": job.matchID, "ok": applied == len(results)}
	if job.batch {
		payload["type"] = "command.batch.ack"
		payload["results"] = results
		payload["applied"] = applied
		payload["failed"] = len(results) - applied
	} else if len(results) == 1 {
		payload["reason"] = results[0].Reason
	}
	if applied != len(results) && job.batch {
		if reason := d.reason(job); reason != "" {
			payload["reason"] = reason
		}
	}
	if job.batch {
		d.mu.Lock()
		d.agent.mu.Lock()
		if job.epoch == d.epoch && d.ctx.Err() == nil && d.agent.currentWS == d.ws &&
			d.agent.state.AssignmentID == job.assignmentID && d.agent.state.MatchID == job.matchID {
			d.agent.state.LastCommand = "vMix batch"
			d.agent.state.LastCommandAt = time.Now()
			d.agent.state.LastCommandOK = applied == len(results)
			d.agent.state.LastCommandDetail = ""
			if applied != len(results) {
				d.agent.state.LastCommandDetail = "Some commands failed"
			}
		}
		d.agent.mu.Unlock()
		d.mu.Unlock()
	}
	if job.bulk {
		reason := "ok"
		for _, row := range results {
			if !row.OK {
				reason = row.Reason
				break
			}
		}
		d.agent.logf("Mapping batch: %d commands, confirmed %d, failed %d, reason %s", len(results), applied, len(results)-applied, reason)
	} else {
		for index, command := range job.commands {
			reason := results[index].Reason
			if results[index].OK {
				reason = "ok"
			}
			d.agent.logf("vMix control: %s · %s · %s · %s", dispatchLogTarget(dispatchFunction(command)), dispatchLogTarget(stringValue(command["Input"])), dispatchLogTarget(stringValue(command["SelectedName"])), reason)
		}
	}
	// Recheck immediately before ACK construction leaves this job. The original
	// captured scope remains attached, so a late packet cannot acknowledge a new
	// assignment. A deadline still preserves genuinely completed positional rows.
	if reason := d.reason(job); reason != "" && reason != "mapping_deadline_exceeded" {
		for i := range results {
			results[i] = dispatchResult{Reason: reason}
		}
		payload["ok"] = false
		payload["reason"] = reason
		if job.batch {
			payload["applied"] = 0
			payload["failed"] = len(results)
		}
	}
	_ = d.write(payload)
}

func dispatchLogTarget(value string) string {
	if strings.Contains(value, "://") {
		return "<URL>"
	}
	value = strings.ReplaceAll(strings.ReplaceAll(value, "\r", " "), "\n", " ")
	if len(value) > 100 {
		return value[:100]
	}
	return value
}

func (d *commandDispatcher) write(payload map[string]any) error {
	d.agent.mu.RLock()
	current := d.agent.currentWS == d.ws && d.ctx.Err() == nil
	d.agent.mu.RUnlock()
	if !current {
		return nil
	}
	if err := d.ws.WriteJSON(payload); err != nil {
		d.cancel()
		_ = d.ws.Close()
		return err
	}
	return nil
}