perf(presence): 联系人列表与在线广播 presence 批量化（Redis pipeline 单次往返）

hub-server/internal/app/events.go

@@ -670,8 +670,14 @@ func (a *App) broadcastOnlineStatus(ctx context.Context, userID string, online b
 	}
 
 	frame := ws.NewFrame(eventType, map[string]string{"user_id": userID})
+	// One pipelined presence round trip for all friends instead of one per
+	// friend (#2154 perf lane); errors degrade to no fanout.
+	onlineSet, err := a.CacheClient.AreOnline(ctx, friendIDs)
+	if err != nil {
+		return
+	}
 	for _, friendID := range friendIDs {
-		if online, _ := a.CacheClient.IsOnline(ctx, friendID); online {
+		if onlineSet[friendID] {
 			a.mgr.PushToUser(friendID, frame)
 		}
 	}

hub-server/internal/app/events_test.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/agenthub/hub-server/internal/cache"
 	"github.com/agenthub/hub-server/internal/config"
+	"github.com/agenthub/hub-server/internal/service/contact"
 	"github.com/agenthub/hub-server/internal/service/session"
 	"github.com/agenthub/hub-server/internal/ws"
 )
@@ -160,3 +161,60 @@ func TestHandleTypingFrameRejectsNonMember(t *testing.T) {
 	require.EqualValues(t, 1, queryCount.Load(), "admission check resolves members once")
 	require.NoError(t, mock.ExpectationsWereMet())
 }
+
+// TestBroadcastOnlineStatusUsesBatchedPresence proves the online/offline
+// fanout resolves friend presence with one pipelined AreOnline call instead
+// of one IsOnline round trip per friend (#2154 perf lane).
+func TestBroadcastOnlineStatusUsesBatchedPresence(t *testing.T) {
+	sqlDB, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sqlDB.Close() })
+
+	gormDB, err := gorm.Open(postgres.New(postgres.Config{
+		Conn:                 sqlDB,
+		PreferSimpleProtocol: true,
+	}), &gorm.Config{
+		DisableAutomaticPing: true,
+		Logger:               gormlogger.Default.LogMode(gormlogger.Silent),
+	})
+	require.NoError(t, err)
+
+	rows := sqlmock.NewRows([]string{"friend_id"}).
+		AddRow("friend-online").
+		AddRow("friend-offline")
+	mock.ExpectQuery(`FROM "friendships"`).WillReturnRows(rows)
+
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = rdb.Close() })
+	cacheClient := cache.NewClient(rdb)
+
+	ctx := context.Background()
+	require.NoError(t, cacheClient.SetRoute(ctx, "friend-online", "desktop", "conn-x"))
+
+	mgr := ws.NewManager()
+	onlinePeer := &ws.Conn{Send: make(chan []byte, 4)}
+	require.NoError(t, mgr.Register(onlinePeer))
+	mgr.SetAuth(onlinePeer.ID, "friend-online", "desktop", "dev-on")
+	offlinePeer := &ws.Conn{Send: make(chan []byte, 4)}
+	require.NoError(t, mgr.Register(offlinePeer))
+	mgr.SetAuth(offlinePeer.ID, "friend-offline", "desktop", "dev-off")
+
+	a := &App{
+		CacheClient:    cacheClient,
+		ContactService: contact.NewService(gormDB, nil, cacheClient),
+		mgr:            mgr,
+	}
+	a.broadcastOnlineStatus(ctx, "user-1", true)
+
+	select {
+	case raw := <-onlinePeer.Send:
+		var f ws.Frame
+		require.NoError(t, json.Unmarshal(raw, &f))
+		require.Equal(t, ws.TypeDeviceOnline, f.Type)
+	case <-time.After(time.Second):
+		t.Fatal("expected the online friend to receive the presence frame")
+	}
+	require.Empty(t, offlinePeer.Send, "offline friend must not receive the presence frame")
+	require.NoError(t, mock.ExpectationsWereMet())
+}

hub-server/internal/cache/client_noop.go

@@ -16,8 +16,11 @@ import (
 // *Client — passing nil will panic.
 type NoOpCache struct{}
 
-func (NoOpCache) Invalidate(ctx context.Context, keys ...string) error                   { return nil }
-func (NoOpCache) IsOnline(ctx context.Context, userID string) (bool, error)              { return false, nil }
+func (NoOpCache) Invalidate(ctx context.Context, keys ...string) error      { return nil }
+func (NoOpCache) IsOnline(ctx context.Context, userID string) (bool, error) { return false, nil }
+func (NoOpCache) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) {
+	return map[string]bool{}, nil
+}
 func (NoOpCache) InitSeqIfAbsent(ctx context.Context, sessionID string, seq int64) error { return nil }
 func (NoOpCache) SetSeq(ctx context.Context, sessionID string, seq int64) error          { return nil }
 func (NoOpCache) AllocateSeq(ctx context.Context, sessionID string) (int64, error) {

hub-server/internal/cache/client_routes.go

@@ -93,6 +93,38 @@ func (c *Client) IsOnline(ctx context.Context, userID string) (bool, error) {
 	return n > 0, nil
 }
 
+// AreOnline reports, for every given user, whether at least one active
+// device route exists. The per-user HLEN commands are bundled into a single
+// Redis pipeline round trip instead of N sequential ones — ListContacts and
+// online-status fanout previously paid one round trip per friend (#2154
+// perf lane). Users missing from the result map are offline; a pipeline
+// error is returned to the caller (callers today ignore it and treat the
+// batch as offline, matching the per-item IsOnline convention).
+func (c *Client) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) {
+	online := make(map[string]bool, len(userIDs))
+	if len(userIDs) == 0 {
+		return online, nil
+	}
+	pipe := c.rdb.Pipeline()
+	cmds := make([]*redis.IntCmd, 0, len(userIDs))
+	ids := make([]string, 0, len(userIDs))
+	for _, userID := range userIDs {
+		if _, dup := online[userID]; dup {
+			continue
+		}
+		online[userID] = false
+		cmds = append(cmds, pipe.HLen(ctx, routeKey(userID)))
+		ids = append(ids, userID)
+	}
+	if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil {
+		return nil, err
+	}
+	for i, cmd := range cmds {
+		online[ids[i]] = cmd.Val() > 0
+	}
+	return online, nil
+}
+
 // GetAllRoutes returns all device routes for a user.
 func (c *Client) GetAllRoutes(ctx context.Context, userID string) (map[string]string, error) {
 	return c.rdb.HGetAll(ctx, routeKey(userID)).Result()

hub-server/internal/cache/client_test.go

@@ -377,6 +377,37 @@ func TestIsOnline(t *testing.T) {
 	assert.False(t, online)
 }
 
+// TestAreOnline covers the batched presence lookup (#2154 perf lane):
+// one pipelined round trip replaces one HLEN per user.
+func TestAreOnline(t *testing.T) {
+	c, _ := testClient(t)
+	ctx := context.Background()
+
+	// Empty input returns an empty map.
+	onlineSet, err := c.AreOnline(ctx, nil)
+	require.NoError(t, err)
+	assert.Empty(t, onlineSet)
+
+	// friend-a has a route (online), friend-b has none (offline).
+	require.NoError(t, c.SetRoute(ctx, "friend-a", "desktop", "conn-1"))
+	onlineSet, err = c.AreOnline(ctx, []string{"friend-a", "friend-b"})
+	require.NoError(t, err)
+	assert.True(t, onlineSet["friend-a"])
+	assert.False(t, onlineSet["friend-b"])
+
+	// Duplicate IDs are deduplicated.
+	onlineSet, err = c.AreOnline(ctx, []string{"friend-a", "friend-a"})
+	require.NoError(t, err)
+	assert.Len(t, onlineSet, 1)
+	assert.True(t, onlineSet["friend-a"])
+
+	// Route removal flips presence back to offline.
+	require.NoError(t, c.DeleteRoute(ctx, "friend-a", "desktop"))
+	onlineSet, err = c.AreOnline(ctx, []string{"friend-a"})
+	require.NoError(t, err)
+	assert.False(t, onlineSet["friend-a"])
+}
+
 func TestGetAllRoutes(t *testing.T) {
 	c, _ := testClient(t)
 	ctx := context.Background()

hub-server/internal/service/contact/service.go

@@ -24,6 +24,8 @@ type Bus interface {
 type Cache interface {
 	Invalidate(ctx context.Context, keys ...string) error
 	IsOnline(ctx context.Context, userID string) (bool, error)
+	// AreOnline batches presence lookups in one round trip (#2154 perf lane).
+	AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error)
 }
 
 // Service owns contact/friendship orchestration: user search, friend request
@@ -310,13 +312,18 @@ func (s *Service) ListContacts(ctx context.Context, userID string) ([]ContactInf
 		return nil, err
 	}
 
+	// Presence in one pipelined round trip instead of one per friend (#2154
+	// perf lane); errors degrade to all-offline, matching the previous
+	// per-item error swallowing.
+	onlineSet, _ := resolveCache(s.cacheClient).AreOnline(ctx, friendIDs)
+
 	result := make([]ContactInfo, 0, len(friends))
 	for _, f := range friends {
 		friend, ok := users[f.FriendID]
 		if !ok {
 			continue
 		}
-		online, _ := resolveCache(s.cacheClient).IsOnline(ctx, friend.ID)
+		online := onlineSet[friend.ID]
 		result = append(result, ContactInfo{
 			UserID:    friend.ID,
 			Username:  friend.Username,

hub-server/internal/service/contact/service_test.go

@@ -24,8 +24,10 @@ import (
 
 // mockContactCache implements Cache for testing.
 type mockContactCache struct {
-	invalidated []string
-	online      map[string]bool
+	invalidated    []string
+	online         map[string]bool
+	isOnlineCalls  int
+	areOnlineCalls [][]string
 }
 
 func (m *mockContactCache) Invalidate(ctx context.Context, keys ...string) error {
@@ -34,12 +36,24 @@ func (m *mockContactCache) Invalidate(ctx context.Context, keys ...string) error
 }
 
 func (m *mockContactCache) IsOnline(ctx context.Context, userID string) (bool, error) {
+	m.isOnlineCalls++
 	if m.online == nil {
 		return false, nil
 	}
 	return m.online[userID], nil
 }
 
+func (m *mockContactCache) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) {
+	m.areOnlineCalls = append(m.areOnlineCalls, append([]string(nil), userIDs...))
+	out := make(map[string]bool, len(userIDs))
+	for _, id := range userIDs {
+		if m.online != nil {
+			out[id] = m.online[id]
+		}
+	}
+	return out, nil
+}
+
 // recordingContactBus is a Bus test double that records Publish calls.
 type recordingContactBus struct {
 	events []bus.Event
@@ -789,6 +803,42 @@ func TestListContacts_BatchesFriendUserLookup(t *testing.T) {
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
+func TestListContacts_BatchesPresenceLookups(t *testing.T) {
+	db, mock, sqlDB := newMockDBContact(t)
+	defer sqlDB.Close()
+
+	mock.ExpectQuery(sqlcFriendshipsByUser).
+		WithArgs("user-1", model.StatusAccepted, 500).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "user_id", "friend_id", "status", "remark"}).
+			AddRow("f-1", "user-1", "friend-a", model.StatusAccepted, "A").
+			AddRow("f-2", "user-1", "friend-b", model.StatusAccepted, "B").
+			AddRow("f-3", "user-1", "friend-c", model.StatusAccepted, "C"))
+
+	mock.ExpectQuery(sqlcUsersByIDs).
+		WithArgs("friend-a", "friend-b", "friend-c").
+		WillReturnRows(sqlmock.NewRows([]string{"id", "username", "password_hash", "nickname", "avatar_url"}).
+			AddRow("friend-a", "friendA", "hash-a", "Friend A", "").
+			AddRow("friend-b", "friendB", "hash-b", "Friend B", "").
+			AddRow("friend-c", "friendC", "hash-c", "Friend C", ""))
+
+	mc := &mockContactCache{online: map[string]bool{"friend-b": true}}
+	svc := NewService(db, nil, mc)
+	contacts, err := svc.ListContacts(context.Background(), "user-1")
+	require.NoError(t, err)
+	require.Len(t, contacts, 3)
+
+	// Presence resolves in exactly one batched call covering all friends; the
+	// per-user IsOnline path must not be used anymore (#2154 perf lane).
+	require.Len(t, mc.areOnlineCalls, 1)
+	assert.ElementsMatch(t, []string{"friend-a", "friend-b", "friend-c"}, mc.areOnlineCalls[0])
+	assert.Zero(t, mc.isOnlineCalls)
+
+	assert.False(t, contacts[0].Online)
+	assert.True(t, contacts[1].Online)
+	assert.False(t, contacts[2].Online)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
 // ==================== UpdateRemark ====================
 
 func TestUpdateRemark(t *testing.T) {