feat(multiplayer): publish allocation progress events

server/store/allocation_match_sql.go

@@ -39,8 +39,8 @@ WHERE mp.match_id = $1
 ORDER BY q.client_build`
 
 const BindAllocatedMatchParticipantsSQL = `WITH bound AS (
-    UPDATE matches
-    SET server_id = $3
+	UPDATE matches
+	SET server_id = $3, revision = revision + 1
     WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL
       AND EXISTS (
           SELECT 1 FROM allocations
@@ -58,7 +58,7 @@ const BindAllocatedMatchParticipantsSQL = `WITH bound AS (
     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)`
+	SELECT (SELECT count(*) FROM participants), (SELECT count(*) FROM advanced), COALESCE((SELECT revision FROM bound), -1)`
 
 const ReleaseAllocatedMatchClaimSQL = `UPDATE matches
 SET allocation_id = NULL, allocation_claimed_at = NULL
@@ -245,18 +245,47 @@ func ClaimAllocatingMatch(ctx context.Context, db *sql.DB, transport string, now
 }
 
 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 {
+	if db == nil || allocation.MatchID == "" || allocation.AllocationID == "" || allocation.ServerID == "" || allocation.State != domain.ServerAllocated || allocation.AllocatedAt.IsZero() {
 		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 {
+		var revision int64
+		if err := tx.QueryRowContext(ctx, BindAllocatedMatchParticipantsSQL, allocation.MatchID, allocation.AllocationID, allocation.ServerID).Scan(&participants, &advanced, &revision); err != nil {
 			return err
 		}
 		if participants == 0 || participants != advanced {
 			return domain.ErrConflict
 		}
-		return nil
+		rows, err := tx.QueryContext(ctx, serverRegistrationParticipantIDsSQL, allocation.MatchID)
+		if err != nil {
+			return err
+		}
+		playerIDs := make([]string, 0, participants)
+		for rows.Next() {
+			var playerID string
+			if err := rows.Scan(&playerID); err != nil {
+				rows.Close()
+				return err
+			}
+			playerIDs = append(playerIDs, playerID)
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return err
+		}
+		if err := rows.Close(); err != nil {
+			return err
+		}
+		payload, err := json.Marshal(map[string]any{
+			"event": "state_changed", "revision": revision, "resource_id": allocation.MatchID,
+			"occurred_at": allocation.AllocatedAt, "state": string(domain.Allocating), "match_id": allocation.MatchID, "player_ids": playerIDs,
+		})
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, serverRegistrationOutboxSQL, fmt.Sprintf("match:%s:%d", allocation.MatchID, revision), allocation.MatchID, revision, payload)
+		return err
 	})
 }
 

server/store/allocation_match_sql_test.go

@@ -11,7 +11,7 @@ func TestAllocationMatchClaimSQLFencesConcurrentWorkers(t *testing.T) {
 	checks := map[string][]string{
 		ClaimAllocatingMatchSQL:                {"FOR UPDATE SKIP LOCKED", "allocation_id = 'allocation-' || candidate.match_id", "allocation_claimed_at <= $1", "ORDER BY created_at, match_id", "m.playlist"},
 		AllocatingMatchBuildSQL:                {"match_participants", "queue_tickets", "ORDER BY q.client_build"},
-		BindAllocatedMatchParticipantsSQL:      {"allocation_id = $2", "server_id IS NULL", "SET server_id = $3", "FROM allocations", "state = 'ALLOCATING'", "revision = revision + 1"},
+		BindAllocatedMatchParticipantsSQL:      {"allocation_id = $2", "server_id IS NULL", "SET server_id = $3", "FROM allocations", "state = 'ALLOCATING'", "revision = revision + 1", "SELECT revision FROM bound"},
 		ReleaseAllocatedMatchClaimSQL:          {"allocation_id = $2", "allocation_id = NULL", "allocation_claimed_at = NULL"},
 		AdvanceServerRegistrationSQL:           {"state = $4", "protocol_version = $7", "ASSIGNMENT_READY", "revision = revision + 1"},
 		ServerRegistrationIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"},

server/store/postgres_integration_test.go

@@ -310,6 +310,14 @@ func TestPostgreSQLAllocationMatchClaimLeaseAndBindFence(t *testing.T) {
 	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'allocation-match-ticket-%' AND state = 'ALLOCATING'`).Scan(&allocatingTickets); err != nil || allocatingTickets != 2 {
 		t.Fatalf("allocating tickets=%d err=%v", allocatingTickets, err)
 	}
+	var eventType string
+	var eventPayload []byte
+	if err := db.QueryRowContext(ctx, `SELECT event_type, payload FROM outbox WHERE aggregate_id = 'allocation-match' AND event_type = 'state_changed'`).Scan(&eventType, &eventPayload); err != nil {
+		t.Fatalf("allocation outbox event: %v", err)
+	}
+	if eventType != "state_changed" || !strings.Contains(string(eventPayload), `"state":"ALLOCATING"`) || !strings.Contains(string(eventPayload), `"allocation-match-a"`) || !strings.Contains(string(eventPayload), `"allocation-match-b"`) {
+		t.Fatalf("allocation outbox event = %s", eventPayload)
+	}
 	if _, found, err := ClaimAllocatingMatch(ctx, db, "enet", now.Add(2*time.Second)); err != nil || found {
 		t.Fatalf("bound match re-claimed found=%t err=%v", found, err)
 	}