From b4135604556b6a0b868119d547c561639f4b226f Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 23 Sep 2026 10:35:06 +0800 Subject: [PATCH] fix tenant group topic and member cache --- server/db/mysql/tenant_groups.go | 4 + server/im_http.go | 30 +++--- server/store/group_message_runtime_test.go | 6 +- server/store/store.go | 112 +++++++++++++++++++-- 4 files changed, 125 insertions(+), 27 deletions(-) diff --git a/server/db/mysql/tenant_groups.go b/server/db/mysql/tenant_groups.go index 40a53b8..cf6efd9 100644 --- a/server/db/mysql/tenant_groups.go +++ b/server/db/mysql/tenant_groups.go @@ -1226,6 +1226,10 @@ func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, if topic == nil || settings == nil || topic.Id == "" || settings.Topic != topic.Id || owner.IsZero() { return t.ErrMalformed } + topic.TenantID = id + topic.Owner = owner.String() + settings.TenantID = id + settings.OwnerUserID = owner ctx, cancel := a.getContextForTx() if cancel != nil { defer cancel() diff --git a/server/im_http.go b/server/im_http.go index 6f04430..10f2d76 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3407,16 +3407,17 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) { eventID := stableIMEventID("message", topic, seq, updated.Version, "recalled") if err = enqueueIMBusinessEvent(principal.TenantID, eventID, "topic", topic, principal.Uid, "message.recalled", map[string]any{"topic": topic, "seq": seq, "version": updated.Version}); err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.recalled"); err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return + logs.Warn.Printf("im message recall: failed to enqueue business event tenant=%d topic=%s seq=%d version=%d: %v", principal.TenantID, topic, seq, updated.Version, err) + } else if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.recalled"); err != nil { + logs.Warn.Printf("im message recall: failed to enqueue p2p mutation events tenant=%d topic=%s seq=%d version=%d: %v", principal.TenantID, topic, seq, updated.Version, err) } } broadcastIMMessageRecalled(principal.TenantID, updated.Topic, []int64{seq}, principal.Uid) - writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled")) + payload := messageMutationPayload(updated, "recalled") + if err != nil { + payload["event_status"] = "enqueue_failed" + } + writeIMSuccess(wrt, req, payload) } func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) { @@ -3554,15 +3555,16 @@ func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) { eventID := stableIMEventID("message", topic, seq, updated.Version, "edited") if err = enqueueIMBusinessEvent(principal.TenantID, eventID, "topic", topic, principal.Uid, "message.edited", map[string]any{"topic": topic, "seq": seq, "version": updated.Version}); err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return - } - if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.edited"); err != nil { - writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) - return + logs.Warn.Printf("im message edit: failed to enqueue business event tenant=%d topic=%s seq=%d version=%d: %v", principal.TenantID, topic, seq, updated.Version, err) + } else if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.edited"); err != nil { + logs.Warn.Printf("im message edit: failed to enqueue p2p mutation events tenant=%d topic=%s seq=%d version=%d: %v", principal.TenantID, topic, seq, updated.Version, err) } } - writeIMSuccess(wrt, req, messageMutationPayload(updated, "edited")) + payload := messageMutationPayload(updated, "edited") + if err != nil { + payload["event_status"] = "enqueue_failed" + } + writeIMSuccess(wrt, req, payload) } func serveIMConversationContext(wrt http.ResponseWriter, req *http.Request) { diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 0136d7f..a334358 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -282,16 +282,16 @@ func TestGroupMemberPageCacheRoundTripAndInvalidate(t *testing.T) { t.Fatalf("cache key must be topic-prefixed for invalidation: %q", key) } CacheGroupMemberPage(7, key, groupMemberPageCache{ - Items: []types.GroupMember{{ + Items: []groupMemberCache{groupMemberCacheFromType(&types.GroupMember{ ID: 43, Topic: "grpAlpha", UserID: types.Uid(100), Role: types.GroupMemberRoleMember, - }}, + })}, NextCursor: "43", }) cached, ok := GetCachedGroupMemberPage(7, key) - if !ok || cached.NextCursor != "43" || len(cached.Items) != 1 || cached.Items[0].UserID != types.Uid(100) { + if !ok || cached.NextCursor != "43" || len(cached.Items) != 1 || cached.Items[0].UserID != types.Uid(100).UserId() { t.Fatalf("unexpected cached page: ok=%v page=%#v", ok, cached) } diff --git a/server/store/store.go b/server/store/store.go index 91e8fe1..44d3f27 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -141,13 +141,13 @@ const ( cacheNamespaceTopic = "topic" cacheNamespaceSub = "sub" cacheNamespaceGroupSettings = "group_settings" - cacheNamespaceGroupMember = "group_member" + cacheNamespaceGroupMember = "group_member:v2" cacheNamespaceVerification = "verification" cacheNamespaceActionToken = "action_token" cacheNamespaceFriends = "friends" cacheNamespaceBlocklist = "blocklist" cacheNamespaceUnreadCount = "unread_count" - cacheNamespaceGroupPage = "group_member_page" + cacheNamespaceGroupPage = "group_member_page:v2" ) // TenantsPersistenceInterface defines tenant registry lookups used before authentication. @@ -1986,7 +1986,7 @@ func (groupBusinessMapper) UpsertMember(tenantID types.TenantID, member *types.G } member, err = tenantAdp.TenantGroupMemberUpsert(tenantID, member) if err == nil && member != nil { - _ = CacheObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(member.Topic, member.UserID), member) + _ = CacheObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(member.Topic, member.UserID), groupMemberCacheFromType(member)) } return member, err } @@ -1997,11 +1997,11 @@ func (groupBusinessMapper) GetMember(tenantID types.TenantID, topic string, uid } topic = strings.TrimSpace(topic) cacheID := groupMemberCacheID(topic, uid) - var cached types.GroupMember + var cached groupMemberCache if ok, err := GetCachedObject(tenantID, cacheNamespaceGroupMember, cacheID, &cached); err != nil { logs.Warn.Printf("cache group member[%s]: read failed: %v", cacheID, err) } else if ok { - return &cached, nil + return cached.toType(), nil } tenantAdp, err := tenantBusinessAdapter(tenantID) if err != nil { @@ -2009,7 +2009,7 @@ func (groupBusinessMapper) GetMember(tenantID types.TenantID, topic string, uid } member, err := tenantAdp.TenantGroupMemberGet(tenantID, topic, uid) if err == nil && member != nil { - if cacheErr := CacheObject(tenantID, cacheNamespaceGroupMember, cacheID, member); cacheErr != nil { + if cacheErr := CacheObject(tenantID, cacheNamespaceGroupMember, cacheID, groupMemberCacheFromType(member)); cacheErr != nil { logs.Warn.Printf("cache group member[%s]: write failed: %v", cacheID, cacheErr) } } @@ -2030,7 +2030,7 @@ func (groupBusinessMapper) ListMembers(tenantID types.TenantID, topic string, op opts.Cursor = strings.TrimSpace(opts.Cursor) cacheKey := groupMemberPageCacheID(topic, opts) if cached, ok := GetCachedGroupMemberPage(tenantID, cacheKey); ok { - return cached.Items, cached.NextCursor, nil + return groupMembersFromCaches(cached.Items), cached.NextCursor, nil } tenantAdp, err := tenantBusinessAdapter(tenantID) if err != nil { @@ -2041,15 +2041,107 @@ func (groupBusinessMapper) ListMembers(tenantID types.TenantID, topic string, op return nil, "", err } CacheGroupMemberPage(tenantID, cacheKey, groupMemberPageCache{ - Items: members, + Items: groupMemberCachesFromTypes(members), NextCursor: nextCursor, }) return members, nextCursor, nil } type groupMemberPageCache struct { - Items []types.GroupMember `json:"items"` - NextCursor string `json:"next_cursor"` + Items []groupMemberCache `json:"items"` + NextCursor string `json:"next_cursor"` +} + +type groupMemberCache struct { + ID int64 `json:"id"` + Topic string `json:"topic"` + UserID string `json:"user_id"` + Role string `json:"role"` + MemberStatus string `json:"member_status"` + Nickname string `json:"nickname,omitempty"` + ModeWant types.AccessMode `json:"mode_want"` + ModeGiven types.AccessMode `json:"mode_given"` + Public any `json:"public,omitempty"` + Trusted any `json:"trusted,omitempty"` + Pinned bool `json:"pinned"` + MuteNotifications bool `json:"mute_notifications"` + JoinSource string `json:"join_source,omitempty"` + JoinedBy string `json:"joined_by,omitempty"` + JoinedAt time.Time `json:"joined_at"` + LeftAt *time.Time `json:"left_at,omitempty"` + RemovedBy string `json:"removed_by,omitempty"` + Version int `json:"version"` + CreatedAt time.Time `json:"created_at,omitempty"` + UpdatedAt time.Time `json:"updated_at,omitempty"` +} + +func groupMemberCachesFromTypes(members []types.GroupMember) []groupMemberCache { + items := make([]groupMemberCache, 0, len(members)) + for i := range members { + items = append(items, groupMemberCacheFromType(&members[i])) + } + return items +} + +func groupMembersFromCaches(items []groupMemberCache) []types.GroupMember { + members := make([]types.GroupMember, 0, len(items)) + for i := range items { + members = append(members, *items[i].toType()) + } + return members +} + +func groupMemberCacheFromType(member *types.GroupMember) groupMemberCache { + if member == nil { + return groupMemberCache{} + } + return groupMemberCache{ + ID: member.ID, + Topic: member.Topic, + UserID: member.UserID.UserId(), + Role: member.Role, + MemberStatus: member.MemberStatus, + Nickname: member.Nickname, + ModeWant: member.ModeWant, + ModeGiven: member.ModeGiven, + Public: member.Public, + Trusted: member.Trusted, + Pinned: member.Pinned, + MuteNotifications: member.MuteNotifications, + JoinSource: member.JoinSource, + JoinedBy: member.JoinedBy.UserId(), + JoinedAt: member.JoinedAt, + LeftAt: member.LeftAt, + RemovedBy: member.RemovedBy.UserId(), + Version: member.Version, + CreatedAt: member.CreatedAt, + UpdatedAt: member.UpdatedAt, + } +} + +func (member groupMemberCache) toType() *types.GroupMember { + return &types.GroupMember{ + ID: member.ID, + Topic: member.Topic, + UserID: types.ParseUserId(member.UserID), + Role: member.Role, + MemberStatus: member.MemberStatus, + Nickname: member.Nickname, + ModeWant: member.ModeWant, + ModeGiven: member.ModeGiven, + Public: member.Public, + Trusted: member.Trusted, + Pinned: member.Pinned, + MuteNotifications: member.MuteNotifications, + JoinSource: member.JoinSource, + JoinedBy: types.ParseUserId(member.JoinedBy), + JoinedAt: member.JoinedAt, + LeftAt: member.LeftAt, + RemovedBy: types.ParseUserId(member.RemovedBy), + Version: member.Version, + CreatedAt: member.CreatedAt, + UpdatedAt: member.UpdatedAt, + } } func groupMemberPageCacheID(topic string, opts types.GroupMemberListOptions) string {