Accept friend requests reactively via social RTA subscription

broadcaster.go

@@ -68,6 +68,8 @@ type Broadcaster struct {
 	failure    error
 	recovering bool
 	acceptWg   sync.WaitGroup
+	// socialWg tracks social subscription cleanup before clients are closed.
+	socialWg sync.WaitGroup
 
 	// lastQuery is the most recent successful target-server query, kept so
 	// query failures fall back to real data instead of failing the update.
@@ -283,7 +285,9 @@ func (b *Broadcaster) Start(ctx context.Context) error {
 			"initial_invite", b.conf.FriendSync.InitialInvite,
 			"expiry_enabled", b.conf.FriendSync.ExpiryEnabled,
 		)
-		go b.friendSyncer().Run(b.ctx)
+		syncer := b.friendSyncer()
+		syncer.Trigger = b.startSocialSubscription(b.conf.XBLClient, b.conf.FriendSync, b.log)
+		go syncer.Run(b.ctx)
 	}
 	b.startSubAccountFriendSync()
 	go b.logSocialSummary()
@@ -341,12 +345,14 @@ func (b *Broadcaster) startSubAccountFriendSync() {
 		if conf == nil {
 			continue
 		}
+		syncLog := b.log.With("sub_account", account.ID)
 		syncer := FriendSyncer{
 			Client:   b.friendClientFor(account.XBLClient),
 			Config:   *conf,
 			History:  b.conf.FriendHistory,
 			Notifier: b.conf.Notifier,
-			Log:      b.log.With("sub_account", account.ID),
+			Trigger:  b.startSocialSubscription(account.XBLClient, conf, syncLog),
+			Log:      syncLog,
 		}
 		if conf.InitialInvite {
 			syncer.Inviter = &subAccountInviter{b: b, id: account.ID}
@@ -1990,6 +1996,8 @@ func (b *Broadcaster) Close() error {
 		return nil
 	}
 	b.cancel()
+	// Release only our social handlers before closing any owned clients.
+	b.socialWg.Wait()
 	var err error
 	if b.listener != nil {
 		err = b.listener.Close()

friend_sync.go

@@ -58,7 +58,12 @@ type FriendSyncer struct {
 	// syncer's friend list. Enable it on exactly one syncer per shared
 	// HistoryStore, or entries maintained by other accounts get dropped.
 	PruneHistory bool
-	Log          *slog.Logger
+	// Trigger, when non-nil, requests an immediate sync pass whenever it
+	// receives a signal. It lets an RTA social subscription accept incoming
+	// friend requests promptly instead of waiting for the next tick. A nil
+	// Trigger leaves the syncer purely periodic.
+	Trigger <-chan struct{}
+	Log     *slog.Logger
 }
 
 const friendListFullBackoff = time.Hour
@@ -520,6 +525,14 @@ func (s FriendSyncer) Run(ctx context.Context) {
 			s.runSync(ctx, &state, false)
 		case <-expiryC:
 			s.runSync(ctx, &state, true)
+		case _, ok := <-s.Trigger:
+			if !ok {
+				// A closed Trigger receives forever; disable it so Run does
+				// not spin and bypass the interval.
+				s.Trigger = nil
+				continue
+			}
+			s.runSync(ctx, &state, false)
 		case <-ctx.Done():
 			return
 		}

go.mod

@@ -3,6 +3,7 @@ module github.com/HashimTheArab/go-mcxboxbroadcast
 go 1.26.2
 
 require (
+	github.com/coder/websocket v1.8.14
 	github.com/df-mc/go-nethernet v1.0.20
 	github.com/df-mc/go-playfab/v2 v2.0.2
 	github.com/df-mc/go-xsapi/v2 v2.0.3
@@ -20,7 +21,6 @@ require (
 
 require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/coder/websocket v1.8.14 // indirect
 	github.com/coreos/go-oidc/v3 v3.17.0 // indirect
 	github.com/creachadair/jrpc2 v1.3.5 // indirect
 	github.com/creachadair/mds v0.26.1 // indirect
@@ -57,4 +57,4 @@ replace (
 	github.com/sandertv/gophertunnel => github.com/hashimthearab/gophertunnel v1.25.3-0.20260902183206-fc666ba2a7ce
 )
 
-replace github.com/df-mc/go-xsapi/v2 => github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911064730-d60df3023755
+replace github.com/df-mc/go-xsapi/v2 => github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911084536-5e753424fe8d

go.sum

@@ -1,5 +1,5 @@
-github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911064730-d60df3023755 h1:Gg8w9/iCHhv9+pxHdi2Bh11cXy3D0k7TiXYSLX++bcA=
-github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911064730-d60df3023755/go.mod h1:Gi/zQG2DFMJOMt4DIjuBuINTnU4YnZINe+RyFyza8oo=
+github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911084536-5e753424fe8d h1:YbLN11lhxU8ruS9AnJ2/6yoiNBq5lXSkABDplTKPv9w=
+github.com/HashimTheArab/go-xsapi/v2 v2.0.0-20260911084536-5e753424fe8d/go.mod h1:Gi/zQG2DFMJOMt4DIjuBuINTnU4YnZINe+RyFyza8oo=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=

social_subscription.go

@@ -0,0 +1,116 @@
+package broadcaster
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/df-mc/go-xsapi/v2"
+	xblsocial "github.com/df-mc/go-xsapi/v2/social"
+)
+
+// socialSubscriber is the part of go-xsapi's social client the broadcaster uses
+// to receive and release RTA relationship events. It is satisfied by
+// [*xblsocial.Client]. Subscribe returns a cleanup function for only that
+// registration, leaving other subscribers on a shared client untouched.
+type socialSubscriber interface {
+	Subscribe(context.Context, xblsocial.SubscriptionHandler) (func(context.Context) error, error)
+}
+
+// reactiveFriendSyncApplicable reports whether an account with the given friend
+// sync configuration benefits from a reactive social subscription. Only
+// auto-follow and auto-unfollow act on social changes, so a subscription is
+// pointless without at least one of them.
+func reactiveFriendSyncApplicable(conf *FriendSyncConfig) bool {
+	return conf != nil && (conf.AutoFollow || conf.AutoUnfollow)
+}
+
+// startSocialSubscription subscribes to the account's RTA social feed so friend
+// requests are accepted (and relationship changes reconciled) reactively rather
+// than only on the sync interval. It returns a channel the account's friend
+// syncer should select on to run an immediate pass, or nil when a reactive
+// subscription is not applicable.
+func (b *Broadcaster) startSocialSubscription(client *xsapi.Client, conf *FriendSyncConfig, log *slog.Logger) <-chan struct{} {
+	if !hasSocialClient(client) || !reactiveFriendSyncApplicable(conf) {
+		return nil
+	}
+	return b.subscribeSocial(client.Social(), log)
+}
+
+// subscribeSocial subscribes sub to the social RTA feed and returns a trigger
+// channel that fires on each event. The subscription's lifetime is bound to the
+// broadcaster: a dedicated goroutine unsubscribes once the broadcaster's context
+// is canceled, and [Broadcaster.Close] waits for it via socialWg. This keeps a
+// caller-provided client (which the broadcaster does not close) from
+// accumulating stale handlers across broadcaster restarts.
+//
+// The subscription is best-effort: on a subscribe failure or a lost subscription
+// the periodic syncer remains the backstop, and the RTA connection re-subscribes
+// automatically after a transient drop.
+func (b *Broadcaster) subscribeSocial(sub socialSubscriber, log *slog.Logger) <-chan struct{} {
+	// Buffered by one so bursts of events collapse into a single pending pass.
+	trigger := make(chan struct{}, 1)
+	handler := friendRequestSubscriptionHandler{trigger: trigger, log: log}
+	b.socialWg.Add(1)
+	go func() {
+		defer b.socialWg.Done()
+		// Subscribe dials RTA lazily; running it here keeps a slow or failing
+		// dial off the start path.
+		unsubscribe, err := sub.Subscribe(b.ctx, handler)
+		if err != nil {
+			log.Warn("subscribe to social rta feed; friend requests will be accepted on the sync interval", "err", err)
+			return
+		}
+		log.Debug("subscribed to social rta feed for reactive friend sync")
+
+		<-b.ctx.Done()
+		// b.ctx is done, so use a fresh context to release the subscription.
+		// The cleanup removes only this registration, so a shared client's other
+		// subscribers keep working.
+		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
+		if err := unsubscribe(ctx); err != nil {
+			log.Debug("unsubscribe social rta feed", "err", err)
+		}
+	}()
+	return trigger
+}
+
+// friendRequestSubscriptionHandler adapts go-xsapi's social RTA subscription to
+// the friend syncer: any relationship change or incoming-friend-request event
+// requests an immediate sync pass through trigger.
+type friendRequestSubscriptionHandler struct {
+	trigger chan<- struct{}
+	log     *slog.Logger
+}
+
+// HandleIncomingFriendRequestCountChange requests a sync pass so newly received
+// friend requests are accepted promptly, mirroring MCXboxBroadcast's reaction to
+// the IncomingFriendRequestCountChanged notification.
+func (h friendRequestSubscriptionHandler) HandleIncomingFriendRequestCountChange(count int) {
+	h.log.Debug("incoming friend request count changed", "count", count)
+	h.signal()
+}
+
+// HandleSocialNotification requests a sync pass when the caller's relationships
+// change (a user added, removed, or updated the caller).
+func (h friendRequestSubscriptionHandler) HandleSocialNotification(typ string, xuids []string) {
+	h.log.Debug("social notification", "type", typ, "xuids", len(xuids))
+	h.signal()
+}
+
+// HandleSubscriptionLost logs the loss. The periodic syncer continues to accept
+// requests, and the RTA connection re-subscribes automatically after transient
+// drops, so no action is taken here.
+func (h friendRequestSubscriptionHandler) HandleSubscriptionLost() {
+	h.log.Warn("social subscription lost; friend requests will be accepted on the sync interval until it is restored")
+}
+
+// signal requests a sync pass without blocking. A full buffer means a pass is
+// already pending, so the event is coalesced into it.
+func (h friendRequestSubscriptionHandler) signal() {
+	select {
+	case h.trigger <- struct{}{}:
+	default:
+	}
+}