CosmicClash/server/api/outbox.go

package api

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

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

// RunProposalOutboxDispatcher delivers committed proposal changes to the
// authenticated WebSocket subscribers. It only reads proposal_changed rows;
// result and other outbox event types remain owned by their own consumers.
// The outbox guarantees after-commit publication into this replica's bounded
// transient hub; WebSocket receipt is deliberately best-effort. Clients use
// owner-scoped periodic REST recovery for correctness across disconnects and
// replicas, so a socket notification is only a latency optimization.
func RunProposalOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) {
	if db == nil || service == nil {
		return
	}
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	dispatcher := store.NewOutboxDispatcher(db, func(deliveryCtx context.Context, event store.OutboxEvent) error {
		return deliverProposalOutboxEvent(deliveryCtx, event, service)
	})
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			events, err := store.ReadUnpublishedProposalOutbox(ctx, db, 100)
			if err != nil {
				continue
			}
			_ = dispatchOutboxEvents(ctx, db, dispatcher, events)
		}
	}
}

// RunResultOutboxDispatcher delivers committed match results as targeted
// COMPLETED state events. It owns only match_completed rows; proposal rows
// remain with RunProposalOutboxDispatcher.
func RunResultOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) {
	if db == nil || service == nil {
		return
	}
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	dispatcher := store.NewOutboxDispatcher(db, func(deliveryCtx context.Context, event store.OutboxEvent) error {
		return deliverResultOutboxEvent(deliveryCtx, db, event, service)
	})
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			events, err := store.ReadUnpublishedResultOutbox(ctx, db, 100)
			if err != nil {
				continue
			}
			_ = dispatchOutboxEvents(ctx, db, dispatcher, events)
		}
	}
}

// RunStateOutboxDispatcher delivers committed allocation/no-show lifecycle
// transitions to each participant without acknowledging proposal or result
// events owned by the other dispatchers.
func RunStateOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) {
	if db == nil || service == nil {
		return
	}
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	dispatcher := store.NewOutboxDispatcher(db, func(deliveryCtx context.Context, event store.OutboxEvent) error {
		return deliverStateOutboxEvent(deliveryCtx, event, service)
	})
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			events, err := store.ReadUnpublishedStateOutbox(ctx, db, 100)
			if err != nil {
				continue
			}
			_ = dispatchOutboxEvents(ctx, db, dispatcher, events)
		}
	}
}

func dispatchOutboxEvents(ctx context.Context, db *sql.DB, dispatcher *store.OutboxDispatcher, events []store.OutboxEvent) error {
	if len(events) == 0 {
		return nil
	}
	// Use the same delivery-before-ack contract as the general dispatcher,
	// while keeping the already-filtered batch from being read a second time.
	//
	// A delivery failure does not abort the batch. Returning here meant one
	// undeliverable payload -- reads are oldest-first -- was retried ahead of
	// every later event of its type on every poll, forever. Instead the failure
	// is counted against that row (dead-lettering it once exhausted) and the
	// batch continues.
	//
	// Ordering within one aggregate is still honoured: once an event for a
	// match fails, its later events are left for a subsequent poll so a client
	// can never observe that match's newer state before its older state. Other
	// aggregates are independent and proceed.
	blocked := make(map[string]struct{})
	var firstErr error
	for _, event := range events {
		if event.EventID == "" {
			return fmt.Errorf("outbox event has no ID")
		}
		if _, skip := blocked[event.AggregateID]; skip {
			continue
		}
		if err := dispatcher.Deliver(ctx, event); err != nil {
			blocked[event.AggregateID] = struct{}{}
			if firstErr == nil {
				firstErr = err
			}
			if db != nil {
				if _, failErr := store.RecordOutboxDeliveryFailure(ctx, db, event.EventID, err, time.Now().UTC()); failErr != nil {
					return failErr
				}
			}
			continue
		}
		if err := dispatcher.Ack(ctx, event.EventID, time.Now().UTC()); err != nil {
			// An ack failure is a database problem, not a payload problem;
			// stop rather than counting it against the event.
			return err
		}
	}
	return firstErr
}

func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error {
	var envelope struct {
		Event      string    `json:"event"`
		Revision   uint64    `json:"revision"`
		ResourceID string    `json:"resource_id"`
		OccurredAt time.Time `json:"occurred_at"`
		State      string    `json:"state"`
		PlayerIDs  []string  `json:"player_ids"`
	}
	if err := json.Unmarshal(event.Payload, &envelope); err != nil {
		return fmt.Errorf("decode proposal outbox event: %w", err)
	}
	if envelope.Event != "proposal_changed" || envelope.ResourceID == "" || len(envelope.PlayerIDs) == 0 {
		return fmt.Errorf("invalid proposal outbox event")
	}
	for _, playerID := range envelope.PlayerIDs {
		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
			Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID,
			OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID,
		}); err != nil {
			return err
		}
	}
	return nil
}

func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.OutboxEvent, service *Service) error {
	if event.EventType != "match_completed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 {
		return fmt.Errorf("invalid result outbox event")
	}
	var payload map[string]any
	if err := json.Unmarshal(event.Payload, &payload); err != nil || payload == nil {
		return fmt.Errorf("decode result outbox event: %w", err)
	}
	players, err := store.ReadMatchParticipantIDs(ctx, db, event.AggregateID)
	if err != nil {
		return err
	}
	if len(players) == 0 {
		return fmt.Errorf("result outbox event has no participants")
	}
	for _, playerID := range players {
		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
			Event: "state_changed", Revision: event.Revision, ResourceID: event.AggregateID,
			OccurredAt: event.CreatedAt, State: "COMPLETED", MatchID: event.AggregateID,
			PlayerID: playerID,
		}); err != nil {
			return err
		}
	}
	return nil
}

func deliverStateOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error {
	if event.EventType != "state_changed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 {
		return fmt.Errorf("invalid state outbox event")
	}
	var envelope struct {
		Event      string    `json:"event"`
		Revision   uint64    `json:"revision"`
		ResourceID string    `json:"resource_id"`
		OccurredAt time.Time `json:"occurred_at"`
		State      string    `json:"state"`
		MatchID    string    `json:"match_id"`
		PlayerIDs  []string  `json:"player_ids"`
	}
	if err := json.Unmarshal(event.Payload, &envelope); err != nil {
		return fmt.Errorf("decode state outbox event: %w", err)
	}
	if envelope.Event != "state_changed" || envelope.ResourceID != event.AggregateID || envelope.Revision != event.Revision || envelope.State == "" || len(envelope.PlayerIDs) == 0 {
		return fmt.Errorf("invalid state outbox payload")
	}
	for _, playerID := range envelope.PlayerIDs {
		if playerID == "" {
			return fmt.Errorf("state outbox event has empty participant")
		}
		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, MatchID: envelope.MatchID, PlayerID: playerID}); err != nil {
			return err
		}
	}
	return nil
}