diff --git a/.gitignore b/.gitignore index 843e01f..7f3801f 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,9 @@ # IDE workspace files. .idea/ +# Trae workspace: AI test scripts, temp files, docs. +/.trae/ + # Local Go build output. /server/server /.codex-gocache/ diff --git a/server/init_topic.go b/server/init_topic.go index 4ac0eaa..f73db82 100644 --- a/server/init_topic.go +++ b/server/init_topic.go @@ -639,6 +639,12 @@ func initTopicGrp(t *Topic) error { return types.ErrTopicNotFound } + // A dissolved group must not be loaded or re-subscribed to. Otherwise the + // group could be restored by any former member. + if stopic.State == types.StateDeleted { + return types.ErrTopicNotFound + } + if err = t.loadSubscribers(); err != nil { return err } diff --git a/server/store/store.go b/server/store/store.go index e3c982d..92b5390 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1743,6 +1743,11 @@ func (groupBusinessMapper) Dissolve(tenantID types.TenantID, topic string, actor invalidateGroupMemberPages(tenantID, topic) invalidateGroupMemberObjects(tenantID, topic) invalidateTopicSubscriptions(tenantID, topic) + // The dissolve flips im_topics.state to StateDeleted. The cached topic + // object must be dropped too, otherwise a later Topics.Get (e.g. from + // initTopicGrp when a former member re-subscribes) would restore the + // dissolved group from the stale cache. + _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, strings.TrimSpace(topic)) } return settings, err } @@ -2132,6 +2137,9 @@ func (groupBusinessMapper) TransferOwner(tenantID types.TenantID, topic string, _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, uid, false)) _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, uid, true)) } + // im_topics.owner is updated by the transfer; drop the cached topic + // object so a later Topics.Get sees the new owner. + _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) invalidateGroupMemberPages(tenantID, topic) return nil } diff --git a/server/topic.go b/server/topic.go index d7112cd..6db9a76 100644 --- a/server/topic.go +++ b/server/topic.go @@ -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)