fix(multiplayer): dispatch live abandonment lifecycle

server/api/outbox_test.go

@@ -80,3 +80,26 @@ func TestDeliverStateOutboxEventValidatesRevisionAndTargets(t *testing.T) {
 		t.Fatal("revision-mismatched state event accepted")
 	}
 }
+
+func TestDeliverStateOutboxEventRoutesLiveAbandonmentLifecycle(t *testing.T) {
+	service := &Service{}
+	first := service.getEventHub().subscribe("player-a")
+	second := service.getEventHub().subscribe("player-b")
+	defer service.getEventHub().unsubscribe(first)
+	defer service.getEventHub().unsubscribe(second)
+	payload := []byte(`{"event":"state_changed","revision":9,"resource_id":"match_1234567890","occurred_at":"1970-01-01T00:16:40Z","state":"LIVE","match_id":"match_1234567890","player_ids":["player-a","player-b"],"abandoned_player_ids":["player-a"]}`)
+	event := store.OutboxEvent{EventType: "state_changed", AggregateID: "match_1234567890", Revision: 9, Payload: payload}
+	if err := deliverStateOutboxEvent(context.Background(), event, service); err != nil {
+		t.Fatalf("live abandonment event rejected: %v", err)
+	}
+	for playerID, subscriber := range map[string]*eventSubscriber{"player-a": first, "player-b": second} {
+		select {
+		case packet := <-subscriber.queue:
+			if !json.Valid(packet) {
+				t.Fatalf("%s received malformed lifecycle packet %q", playerID, packet)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("%s did not receive live-abandonment lifecycle event", playerID)
+		}
+	}
+}

server/store/live_abandonment_sql.go

@@ -53,9 +53,14 @@ SET revision = revision + 1
 WHERE match_id = $1 AND state = 'LIVE'
 RETURNING revision`
 
+const liveAbandonmentTargetsSQL = `SELECT player_id
+FROM match_participants
+WHERE match_id = $1 AND participation_active
+ORDER BY player_id`
+
 const liveAbandonmentOutboxSQL = `INSERT INTO outbox
     (event_id, aggregate_type, aggregate_id, revision, event_type, payload)
-VALUES ($1, 'match', $2, $3, 'participant_abandoned', $4)`
+VALUES ($1, 'match', $2, $3, 'state_changed', $4)`
 
 // ReconcileLiveAbandonments applies a bounded, durable reconnect-grace sweep.
 // It does not deactivate participants or alter LIVE tickets: an abandonment
@@ -148,7 +153,18 @@ func ApplyLiveAbandonments(ctx context.Context, db *sql.DB, matchID string, now
 		if err := tx.QueryRowContext(ctx, liveAbandonmentRevisionSQL, matchID).Scan(&revision); err != nil {
 			return err
 		}
-		payload, err := json.Marshal(map[string]any{"match_id": matchID, "abandoned_player_ids": abandonmentIDs(planned)})
+		targets, err := loadLiveAbandonmentTargets(ctx, tx, matchID)
+		if err != nil {
+			return err
+		}
+		if len(targets) == 0 {
+			return fmt.Errorf("%w: live match has no active event targets", domain.ErrConflict)
+		}
+		payload, err := json.Marshal(map[string]any{
+			"event": "state_changed", "revision": revision, "resource_id": matchID,
+			"occurred_at": now, "state": domain.Live, "match_id": matchID,
+			"player_ids": targets, "abandoned_player_ids": abandonmentIDs(planned),
+		})
 		if err != nil {
 			return err
 		}
@@ -204,6 +220,23 @@ func loadLiveAbandonmentHistory(ctx context.Context, tx *sql.Tx, participants []
 	return history, nil
 }
 
+func loadLiveAbandonmentTargets(ctx context.Context, tx *sql.Tx, matchID string) ([]string, error) {
+	rows, err := tx.QueryContext(ctx, liveAbandonmentTargetsSQL, matchID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var players []string
+	for rows.Next() {
+		var playerID string
+		if err := rows.Scan(&playerID); err != nil {
+			return nil, err
+		}
+		players = append(players, playerID)
+	}
+	return players, rows.Err()
+}
+
 func abandonmentIDs(abandonments []domain.Abandonment) []string {
 	ids := make([]string, len(abandonments))
 	for i := range abandonments {

server/store/live_abandonment_sql_test.go

@@ -13,7 +13,8 @@ func TestLiveAbandonmentSQLPreservesResultRosterAndReconnectFences(t *testing.T)
 		liveAbandonmentParticipantSQL:  {"SET abandoned_at", "participation_active", "abandoned_at IS NULL", "RETURNING"},
 		liveAbandonmentPenaltySQL:      {"MATCH_ABANDONED", "ON CONFLICT"},
 		liveAbandonmentRevisionSQL:     {"state = 'LIVE'", "revision = revision + 1"},
-		liveAbandonmentOutboxSQL:       {"participant_abandoned", "revision"},
+		liveAbandonmentOutboxSQL:       {"state_changed", "revision"},
+		liveAbandonmentTargetsSQL:      {"participation_active", "ORDER BY player_id"},
 	} {
 		for _, fragment := range fragments {
 			if !strings.Contains(query, fragment) {

server/store/postgres_integration_test.go

@@ -689,7 +689,7 @@ func TestPostgreSQLLiveReconnectGraceExpiryPersistsAbandonmentWithoutReleasingRe
 		t.Fatalf("penalty ends=%v err=%v", endsAt, err)
 	}
 	var outboxCount, revision int
-	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM outbox WHERE aggregate_id = 'live-abandon-match' AND event_type = 'participant_abandoned'`).Scan(&outboxCount); err != nil || outboxCount != 1 {
+	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM outbox WHERE aggregate_id = 'live-abandon-match' AND event_type = 'state_changed'`).Scan(&outboxCount); err != nil || outboxCount != 1 {
 		t.Fatalf("outbox=%d err=%v", outboxCount, err)
 	}
 	if err := db.QueryRowContext(ctx, `SELECT revision FROM matches WHERE match_id = 'live-abandon-match'`).Scan(&revision); err != nil || revision != 1 {