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

fix: invalidate group settings cache on membership changes

Browse Source
This commit is contained in:
richjohnrambo
2026-09-16 01:03:40 +08:00
parent 139681fadb
commit 9626e8ee7c
5 changed files with 393 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/imwsaudit/main.go
+168 -1
View File
@@ -55,6 +55,8 @@ type audit struct {
p2pTopic string
bobTopic string
groupTopic string
groupSeq int
groupFile string
fileURL string
messageID string
}
@@ -330,9 +332,26 @@ func (a *audit) run() {
return "topic=" + topic
})
a.step("WS-21", "A/B 订阅群 Topic", "WS", a.groupTopic, func() string {
if alice == nil || bob == nil || a.groupTopic == "" {
if a.groupTopic == "" {
return "FAIL: websocket client or group missing"
}
// Reconnect before the group-specific checks. A read timeout permanently
// poisons a gorilla/websocket connection and must not cascade into this section.
if alice != nil && alice.conn != nil {
_ = alice.conn.Close()
}
if bob != nil && bob.conn != nil {
_ = bob.conn.Close()
}
var err error
alice, err = a.openLoggedIn("A-group", a.userA, a.tokenA)
if err != nil {
return "FAIL: A reconnect: " + err.Error()
}
bob, err = a.openLoggedIn("B-group", a.userB, a.tokenB)
if err != nil {
return "FAIL: B reconnect: " + err.Error()
}
if err := alice.sub(a.groupTopic); err != nil {
return "FAIL: A sub group: " + err.Error()
}
@@ -350,6 +369,112 @@ func (a *audit) run() {
if err != nil {
return "FAIL: " + err.Error()
}
a.groupSeq = seq
return fmt.Sprintf("seq=%d topic=%s", seq, topic)
})
a.step("WS-22E", "群消息 client_mid 幂等去重", "WS", a.groupTopic, func() string {
if alice == nil {
return "FAIL: websocket client missing"
}
head := map[string]any{"mime": "text/plain", "x-im-client-mid": "dedup-" + suffix}
first, err := alice.pubCtrl(a.groupTopic, head, "群消息幂等测试 "+suffix)
if err != nil {
return "FAIL: first pub: " + err.Error()
}
second, err := alice.pubCtrl(a.groupTopic, head, "群消息幂等测试 "+suffix)
if err != nil {
return "FAIL: duplicate pub: " + err.Error()
}
if first <= 0 || second != first {
return fmt.Sprintf("FAIL: first_seq=%d duplicate_seq=%d", first, second)
}
return fmt.Sprintf("first_seq=%d duplicate_seq=%d", first, second)
})
a.step("WS-22F", "群消息上下文查询", "HTTP", "/v1/im/topics/{topic}/messages/{seq}/context", func() string {
time.Sleep(500 * time.Millisecond)
status, r, err := a.call("GET", fmt.Sprintf("/v1/im/topics/%s/messages/%d/context?before=5&after=5", url.PathEscape(a.groupTopic), a.groupSeq), nil, a.tokenB)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
anchor, count := asInt(r.Data["anchor_seq"]), len(asSlice(r.Data["items"]))
if anchor != a.groupSeq || count == 0 {
return fmt.Sprintf("FAIL: anchor_seq=%d items=%d", anchor, count)
}
return fmt.Sprintf("anchor_seq=%d items=%d", anchor, count)
})
a.step("WS-22G", "群聊天记录搜索", "HTTP", "/v1/im/topics/{topic}/messages/search", func() string {
status, r, err := a.call("GET", "/v1/im/topics/"+url.PathEscape(a.groupTopic)+"/messages/search?q="+url.QueryEscape("WS 群消息")+"&type=text", nil, a.tokenB)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
count := len(asSlice(r.Data["items"]))
if count == 0 {
return "FAIL: items=0"
}
return fmt.Sprintf("items=%d", count)
})
a.step("WS-22H", "群消息发送者撤回", "HTTP", "/v1/im/topics/{topic}/messages/{seq}/recall", func() string {
status, r, err := a.call("POST", fmt.Sprintf("/v1/im/topics/%s/messages/%d/recall", url.PathEscape(a.groupTopic), a.groupSeq), map[string]any{"version": 1, "reason": "WS 群消息撤回联调"}, a.tokenA)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
return fmt.Sprintf("seq=%d status=%s version=%d", asInt(r.Data["seq"]), asString(r.Data["status"]), asInt(r.Data["version"]))
})
a.step("WS-22A", "HTTP 禁言群成员 B", "HTTP", "/v1/im/groups/{topic}/mutes/batch", func() string {
status, r, err := a.call("POST", "/v1/im/groups/"+url.PathEscape(a.groupTopic)+"/mutes/batch", map[string]any{
"user_ids": []string{a.userB}, "reason": "WS 禁言策略联调",
}, a.tokenA)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
return "member B muted"
})
a.step("WS-22B", "被禁言成员 B pub 被拒绝", "WS", a.groupTopic, func() string {
if bob == nil {
return "FAIL: websocket client missing"
}
id := "pub-muted-" + suffix
code, text, err := bob.sendCtrl(id, map[string]any{"pub": map[string]any{
"id": id, "topic": a.groupTopic, "head": map[string]any{"mime": "text/plain"}, "content": "本消息不应发出",
}}, 15*time.Second)
if err != nil {
return "FAIL: " + err.Error()
}
if code != http.StatusForbidden {
return fmt.Sprintf("FAIL: unexpected code=%d text=%s", code, text)
}
return fmt.Sprintf("code=%d text=%s", code, text)
})
a.step("WS-22C", "HTTP 解除群成员 B 禁言", "HTTP", "/v1/im/groups/{topic}/mutes/{user_id}", func() string {
status, r, err := a.call("DELETE", "/v1/im/groups/"+url.PathEscape(a.groupTopic)+"/mutes/"+url.PathEscape(a.userB), nil, a.tokenA)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
return "member B unmuted"
})
a.step("WS-22D", "解除禁言后 B pub，A 收到 data", "WS", a.groupTopic, func() string {
if alice == nil || bob == nil {
return "FAIL: websocket client missing"
}
seq, topic, err := bob.pubWithHeadAndPeer(alice, a.groupTopic, map[string]any{"mime": "text/plain", "x-im-client-mid": "unmute-" + suffix}, "解除禁言消息 "+suffix)
if err != nil {
return "FAIL: " + err.Error()
}
return fmt.Sprintf("seq=%d topic=%s", seq, topic)
})
a.step("WS-23", "get 群 desc/sub/data", "WS", a.groupTopic, func() string {
@@ -369,6 +494,43 @@ func (a *audit) run() {
a.fileURL = fileURL
return "url=" + mask(fileURL)
})
a.step("WS-24A", "群文件关联 OSS 上传文件", "HTTP", "/v1/im/groups/{topic}/files", func() string {
status, r, err := a.call("POST", "/v1/im/groups/"+url.PathEscape(a.groupTopic)+"/files", map[string]any{
"file_id": a.fileURL, "file_name": "imwsaudit.txt", "mime_type": "text/plain", "file_type": "document",
}, a.tokenA)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK {
return fmt.Sprintf("FAIL: status=%d reason=%s", status, r.Reason)
}
a.groupFile = asString(r.Data["group_file_id"])
if a.groupFile == "" {
return "FAIL: missing group_file_id"
}
return "group_file_id=" + a.groupFile
})
a.step("WS-24B", "群文件列表可查询", "HTTP", "/v1/im/groups/{topic}/files", func() string {
status, r, err := a.call("GET", "/v1/im/groups/"+url.PathEscape(a.groupTopic)+"/files?query=imwsaudit&type=document", nil, a.tokenB)
if err != nil {
return "FAIL: " + err.Error()
}
count := len(asSlice(r.Data["items"]))
if status != http.StatusOK || count == 0 {
return fmt.Sprintf("FAIL: status=%d reason=%s items=%d", status, r.Reason, count)
}
return fmt.Sprintf("items=%d", count)
})
a.step("WS-24C", "群主删除群文件关联", "HTTP", "/v1/im/groups/{topic}/files/{group_file_id}", func() string {
status, r, err := a.call("DELETE", "/v1/im/groups/"+url.PathEscape(a.groupTopic)+"/files/"+url.PathEscape(a.groupFile), nil, a.tokenA)
if err != nil {
return "FAIL: " + err.Error()
}
if status != http.StatusOK || asString(r.Data["status"]) != "deleted" {
return fmt.Sprintf("FAIL: status=%d reason=%s file_status=%s", status, r.Reason, asString(r.Data["status"]))
}
return "status=deleted"
})
a.step("WS-25", "WS 文件消息下发", "WS", a.p2pTopic, func() string {
if alice == nil || bob == nil || a.fileURL == "" {
return "FAIL: websocket client or file missing"
@@ -1028,6 +1190,11 @@ func asInt(v any) int {
}
}
func asSlice(v any) []any {
items, _ := v.([]any)
return items
}
func mask(s string) string {
if len(s) <= 8 {
return s