fix(server): 修复群组权限缓存与解散群问题
1. 新增group事件处理逻辑，实时刷新内存中的群组权限与状态 2. 加载topic时过滤已解散的群组，防止被旧成员恢复 3. 补充群成员离开事件的用户ID字段，完善事件处理
This commit is contained in:
Austin
+200
@@ -10,6 +10,7 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sort"
|
||||
"strconv"
|
||||
@@ -656,6 +657,12 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) {
|
||||
if msg.Pres != nil {
|
||||
t.handlePresence(msg)
|
||||
} else if msg.Info != nil || msg.Data != nil {
|
||||
if msg.Info != nil && msg.Info.What == "group" {
|
||||
// Group management event published by the HTTP or Kafka admin
|
||||
// interface: update in-memory topic state (cached permissions,
|
||||
// owner, lifecycle) before broadcasting the event to clients.
|
||||
t.handleGroupInfoEvent(msg.Info)
|
||||
}
|
||||
t.broadcastToSessions(msg)
|
||||
} else {
|
||||
// TODO(gene): maybe remove this panic.
|
||||
@@ -663,6 +670,199 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) {
|
||||
}
|
||||
}
|
||||
|
||||
// flexUID accepts a JSON string or number representation of a user id: the
|
||||
// HTTP interface publishes base32 user id strings, while Kafka admin commands
|
||||
// publish numeric database ids.
|
||||
type flexUID struct {
|
||||
num int64
|
||||
str string
|
||||
isNum bool
|
||||
}
|
||||
|
||||
func (f *flexUID) UnmarshalJSON(b []byte) error {
|
||||
if len(b) == 0 || string(b) == "null" {
|
||||
return nil
|
||||
}
|
||||
if b[0] == '"' {
|
||||
var s string
|
||||
if err := json.Unmarshal(b, &s); err != nil {
|
||||
return err
|
||||
}
|
||||
f.str = s
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal(b, &f.num); err != nil {
|
||||
return err
|
||||
}
|
||||
f.isNum = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f flexUID) uid() types.Uid {
|
||||
if f.isNum {
|
||||
return types.Uid(f.num)
|
||||
}
|
||||
if f.str == "" {
|
||||
return types.ZeroUid
|
||||
}
|
||||
return types.ParseUserId(f.str)
|
||||
}
|
||||
|
||||
// groupEventPayload is a union of the payloads published with group.* realtime
|
||||
// events by the HTTP and Kafka admin interfaces.
|
||||
type groupEventPayload struct {
|
||||
Owner flexUID `json:"owner"`
|
||||
PreviousOwner flexUID `json:"previous_owner"`
|
||||
UserID flexUID `json:"user_id"`
|
||||
Items []struct {
|
||||
UserID flexUID `json:"user_id"`
|
||||
} `json:"items"`
|
||||
Results []struct {
|
||||
UserID flexUID `json:"user_id"`
|
||||
OK bool `json:"ok"`
|
||||
} `json:"results"`
|
||||
}
|
||||
|
||||
// handleGroupInfoEvent applies server-published group management events to the
|
||||
// in-memory topic state: refreshes the cached per-user permissions or shuts the
|
||||
// topic down when the group is dissolved. The event is still broadcast to the
|
||||
// connected clients afterwards by the caller.
|
||||
func (t *Topic) handleGroupInfoEvent(info *MsgServerInfo) {
|
||||
var payload groupEventPayload
|
||||
if len(info.Payload) > 0 {
|
||||
if err := json.Unmarshal(info.Payload, &payload); err != nil {
|
||||
logs.Warn.Printf("topic[%s]: failed to decode group event payload event=%s: %v", t.name, info.Event, err)
|
||||
}
|
||||
}
|
||||
|
||||
actor := types.ParseUserId(info.From)
|
||||
if actor.IsZero() {
|
||||
actor = payload.UserID.uid()
|
||||
}
|
||||
|
||||
switch info.Event {
|
||||
case "group.owner.transferred":
|
||||
newOwner := payload.Owner.uid()
|
||||
prevOwner := payload.PreviousOwner.uid()
|
||||
t.refreshPerUserFromDB(actor, newOwner, prevOwner)
|
||||
if !newOwner.IsZero() && t.owner == prevOwner {
|
||||
t.owner = newOwner
|
||||
}
|
||||
case "group.role.updated", "group.member.updated", "group.member.left", "group.mute.updated":
|
||||
t.refreshPerUserFromDB(actor, groupEventAffectedUids(&payload)...)
|
||||
case "group.dissolved":
|
||||
// The group is dissolved: stop accepting requests, unregister from the
|
||||
// hub and terminate. handleTopicTermination will detach all sessions
|
||||
// and notify the subscribers.
|
||||
t.markPaused(true)
|
||||
if globals.hub != nil {
|
||||
globals.hub.topicDel(types.TopicKey{TenantID: t.tenantID, Topic: t.name})
|
||||
}
|
||||
select {
|
||||
case t.exit <- &shutDown{reason: StopDeleted}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// groupEventAffectedUids extracts the ids of users affected by a group event.
|
||||
func groupEventAffectedUids(p *groupEventPayload) []types.Uid {
|
||||
uids := make([]types.Uid, 0, 1+len(p.Items)+len(p.Results))
|
||||
if uid := p.UserID.uid(); !uid.IsZero() {
|
||||
uids = append(uids, uid)
|
||||
}
|
||||
for _, item := range p.Items {
|
||||
if uid := item.UserID.uid(); !uid.IsZero() {
|
||||
uids = append(uids, uid)
|
||||
}
|
||||
}
|
||||
for _, item := range p.Results {
|
||||
if uid := item.UserID.uid(); !uid.IsZero() {
|
||||
uids = append(uids, uid)
|
||||
}
|
||||
}
|
||||
return uids
|
||||
}
|
||||
|
||||
// refreshPerUserFromDB reloads the cached per-user permissions for the given
|
||||
// users from the database, e.g. after an out-of-band change (HTTP/Kafka) such
|
||||
// as an owner transfer, role update or mute. Users whose subscription is
|
||||
// missing or deleted are dropped from the cache.
|
||||
func (t *Topic) refreshPerUserFromDB(actor types.Uid, uids ...types.Uid) {
|
||||
if t.isProxy {
|
||||
// Proxy topics carry no per-user state; the master node handles it.
|
||||
return
|
||||
}
|
||||
|
||||
changed := false
|
||||
for _, uid := range uids {
|
||||
if uid.IsZero() {
|
||||
continue
|
||||
}
|
||||
if actor.IsZero() {
|
||||
actor = uid
|
||||
}
|
||||
|
||||
sub, err := store.Subs.Get(t.tenantID, t.name, uid, true)
|
||||
if err != nil {
|
||||
logs.Warn.Printf("topic[%s]: failed to reload subscription for user %s: %v", t.name, uid.UserId(), err)
|
||||
continue
|
||||
}
|
||||
|
||||
pud, found := t.perUser[uid]
|
||||
if sub == nil || sub.DeletedAt != nil {
|
||||
// Subscription is gone: notify and detach the user's sessions.
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
oldWant, oldGiven := pud.modeWant, pud.modeGiven
|
||||
t.notifySubChange(uid, actor, false,
|
||||
oldWant, oldGiven, types.ModeUnset, types.ModeUnset, "")
|
||||
t.evictUser(uid, true, "")
|
||||
changed = true
|
||||
continue
|
||||
}
|
||||
|
||||
if !found {
|
||||
// Member added while the topic was live.
|
||||
t.perUser[uid] = perUserData{
|
||||
private: sub.Private,
|
||||
modeWant: sub.ModeWant,
|
||||
modeGiven: sub.ModeGiven,
|
||||
}
|
||||
changed = true
|
||||
t.notifySubChange(uid, actor, false,
|
||||
types.ModeNone, types.ModeNone, sub.ModeWant, sub.ModeGiven, "")
|
||||
continue
|
||||
}
|
||||
|
||||
oldWant, oldGiven := pud.modeWant, pud.modeGiven
|
||||
if oldWant == sub.ModeWant && oldGiven == sub.ModeGiven {
|
||||
continue
|
||||
}
|
||||
pud.modeWant = sub.ModeWant
|
||||
pud.modeGiven = sub.ModeGiven
|
||||
t.perUser[uid] = pud
|
||||
changed = true
|
||||
|
||||
// Mirror the unread count adjustments made on regular mode changes.
|
||||
oldReader := (oldWant & oldGiven).IsReader()
|
||||
newReader := (sub.ModeWant & sub.ModeGiven).IsReader()
|
||||
if oldReader && !newReader {
|
||||
usersUpdateUnread(t.tenantID, uid, pud.readID-t.lastID, true)
|
||||
} else if !oldReader && newReader {
|
||||
usersUpdateUnread(t.tenantID, uid, t.lastID-pud.readID, true)
|
||||
}
|
||||
|
||||
t.notifySubChange(uid, actor, false,
|
||||
oldWant, oldGiven, sub.ModeWant, sub.ModeGiven, "")
|
||||
}
|
||||
|
||||
if changed {
|
||||
t.computePerUserAcsUnion()
|
||||
}
|
||||
}
|
||||
|
||||
// Session subscribed to a topic, created == true if topic was just created and {pres} needs to be announced
|
||||
func (t *Topic) handleSubscription(msg *ClientComMessage) error {
|
||||
asUid := types.ParseUserId(msg.AsUser)
|
||||
|
||||
Reference in New Issue
Block a user