pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix tenant group topic and member cache

Browse Source
This commit is contained in:
richjohnrambo
2026-09-23 10:35:06 +08:00
parent 6303d07580
commit b413560455
4 changed files with 125 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/tenant_groups.go
+4
View File
@@ -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()
server/im_http.go
+16 -14
View File
@@ -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) {
server/store/group_message_runtime_test.go
+3 -3
View File
@@ -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)
}
server/store/store.go
+102 -10
View File
@@ -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 {