From 3386679d45c22bcfaff3953650d3fa268ca7134f Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 18:51:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20=E9=80=82=E9=85=8D=E5=99=A8?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=8E=A5=20WS=20=E5=AE=9E=E6=97=B6=E6=8E=A8?= =?UTF-8?q?=E9=80=81(=E8=90=BD=E5=BA=93=E5=8D=B3=E7=A7=92=E6=8E=A8,=20chan?= =?UTF-8?q?nel=5Fpersisted)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端: persistIsolated 由 _ = full(不广播)改为 notifyPersisted(full); BroadcastPersisted 按 sessionId 分流——main/web: → message_persisted(Web聊天同步), 隔离 sess-xxx → channel_persisted (适配器日志专用, 永不写主会话 web, 红线). payload data:{message:historyMessage}(strip thinking). 前端: ws.ts 新增 channel_persisted case→parseAdminChannelMessage→pushChannelEvent; wsParse.ts 纯函数; useAppStore 新增 channelEvents(最新在前,按id去重,cap200,不持久化); channelLogList 订阅过滤 (单adapter按adapterId/总览取adapterId非空)merge进列表(仅追加、id去重、不整页重载),保留拉历史+10s兜底. Docs/api-client.md §3.4 补 channel_persisted 契约. 验收: 后端 build/vet/test 全绿+windows/darwin交叉编译绿; 前端 npm run build 99模块绿. 注: 偏离=web: 归 message_persisted(保Web副会话同步), 隔离永不带web前缀红线仍守. --- Docs/api-client.md | 8 ++- .../admin/modules/channels/channelLogList.tsx | 16 +++++- frontend/web/src/lib/ws.ts | 10 ++++ frontend/web/src/lib/wsParse.ts | 27 +++++++++- frontend/web/src/store/useAppStore.ts | 15 ++++++ .../services/channelbridge/channelbridge.go | 13 +++-- .../channelbridge/channelbridge_test.go | 21 +++++++- internal/services/channelbridge/persist.go | 19 ++++--- internal/services/gateway/extra_test.go | 49 +++++++++++++++++++ internal/services/gateway/server.go | 25 +++++++--- 10 files changed, 180 insertions(+), 23 deletions(-) diff --git a/Docs/api-client.md b/Docs/api-client.md index 84df15b..23ae5ab 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -143,9 +143,15 @@ ### 3.3 消息持久化增量事件（`message_persisted`，design/16） - **触发**：每一条 user / assistant 消息**落库后**由编排层广播——Web 主会话路径走 `persistUserMessages`/`persistAssistantReply`，**IM 渠道主人私聊**（design/16 §4.4）由 channelbridge 把消息**落主会话 `web`**（带 `platform` 标记）后复用**同一** `message_persisted` 广播（同一 hub 的 `BroadcastPersisted`），保证 Web 端实时镜像 IM 消息。 - **载荷**：`{type:"message_persisted", sessionId, data:{message:{id, sessionId, role, content, mode?, mood?, image?, platform?, deviceId?, senderName?, createdAt}}}`——`role` ∈ `user|assistant`；`id` 为**后端生成的消息 id**（客户端据此去重）；`platform` 为**来源平台**（`web` 或渠道如 `telegram`/`weixin-ilink`/`qq-obv11`）；`deviceId` 为**具体端/客户端标识**（design/16 §4.5，如 `web:chrome:a1b2c3` / `weixin-ilink:conf0:wx_uid`）；`senderName` 为**第三方发送者昵称**（design/16 §4.6，群聊/陌生人消息带，主人私聊为空，`omitempty`）；`thinking` 已剥离（design/16 §6）。 -- **注：群聊/陌生人（非主人）消息**只在**该 IM 适配器日志界面**以消息流显示，**不**经 `message_persisted` 进 Web 主会话（design/16 §4.6 §3-4）；该日志流由 `GET /api/v1/admin/channels/{platform}/messages` 拉取。 +- **注：群聊/陌生人（非主人）消息**只在**该 IM 适配器日志界面**以消息流显示，**不**经 `message_persisted` 进 Web 主会话（design/16 §4.6 §3-4）；该日志流由 `GET /api/v1/admin/channels/{platform}/messages` 拉取，落库后另经 `channel_persisted`（§3.4）实时推送。 - **客户端处理**：仅当 `data.message.sessionId === activeSessionId`（激活会话）时处理；按 `id` upsert——本端已预建的**同 role 乐观气泡**（本地 `uid` 无后端 `id`）被**归一**（`uid` → 后端 `id`），异端无本地气泡则直接追加，两者最终收敛为**同一条 `id` 相同的消息**，「A 发消息、A 与 B 都无需刷新且各出现一次」。 +### 3.4 适配器消息实时事件（`channel_persisted`，design/16） +- **触发**：每一条**群聊/陌生人（非主人 / 隔离 sess-xxx）消息**落库后由 channelbridge `persistIsolated` 广播；该消息**只写自己的隔离会话**，**绝不**写主会话 `web`，`channel_persisted` 仅是事件通知，不是把内容塞进主会话/Web 上下文（design/16 §4.6 §3-4）。 +- **载荷**：`{type:"channel_persisted", sessionId, data:{message:{id, sessionId, role, content, mode?, platform?, deviceId?, adapterId?, senderName?, createdAt}}}`——与 `historyMessage` wire 一致；`adapterId` 为**产生该消息的适配器注册名**（如 `kook:3bef...`，`adapter != ''`）；`sessionId` 为隔离会话 id（`sess-xxx`，**非** `web`）；`role` ∈ `user|assistant`；`senderName` 为第三方昵称。 +- **事件类型分流**（`BroadcastPersisted` 按 `sessionId` 分流）：`sessionId == "web"` 或其 `web:` 前缀 → **`message_persisted`**（Web 聊天实时同步）；**其它**（隔离 `sess-xxx`）→ **`channel_persisted`**（适配器日志专用）。 +- **客户端处理**：Web 聊天**无** `channel_persisted` handler，天然忽略、不进会话；**IM 适配器日志页**订阅该事件，按 `adapterId === 当前适配器`（或总览 `adapterId != ''`）过滤后**仅追加**（按 `id` 去重、不整页重载），首次拉取仍由 `GET /api/v1/admin/channels/messages` + 轮询兜底。 + ## 4. 客户端接入指引（Web / 移动 / 原生 / IM bridge 通用） ### 4.1 一次性接入流程（三步走） diff --git a/frontend/web/src/admin/modules/channels/channelLogList.tsx b/frontend/web/src/admin/modules/channels/channelLogList.tsx index 530c42c..aeb13c6 100644 --- a/frontend/web/src/admin/modules/channels/channelLogList.tsx +++ b/frontend/web/src/admin/modules/channels/channelLogList.tsx @@ -1,6 +1,7 @@ import { adminChannelMessages, adminChannelLogOverview } from "../../../lib/api"; import type { AdminChannelMessage } from "../../../lib/types"; import { useAsyncData } from "../../../lib/useAsyncData"; +import { useAppStore } from "../../../store/useAppStore"; import { Button, Tag } from "../../../components/primitives"; import { Notice } from "../../components"; import { useAutoReload } from "../../hooks"; @@ -48,7 +49,20 @@ export function ChannelLogList({ [adapter, limit], ); useAutoReload(reload, 10000); - const msgs = data ? [...data].reverse() : data; + + // Realtime tail pushed by `channel_persisted` WS frames (design/16 §4.6 §4). The + // store keeps them newest-first, capped at 200 and deduped by id; we filter to the + // surface this list renders (single adapter, or the overview which shows adapter + // messages only), flip to oldest→newest, then append only ids the history page has + // not already covered. This makes a new persisted message appear without a reload; + // the 10s useAutoReload stays as a fallback for events the socket missed. + const channelEvents = useAppStore((s) => s.channelEvents); + const base = data ? [...data].reverse() : []; + const live = channelEvents + .filter((m) => (adapter ? m.adapterId === adapter : m.adapterId)) + .reverse(); + const seen = new Set(base.map((m) => m.id)); + const msgs = [...base, ...live.filter((m) => !seen.has(m.id))]; return (
diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index c176591..0eca450 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -2,6 +2,7 @@ import { useAppStore } from "../store/useAppStore"; import { useModeStore } from "../store/useModeStore"; import { getAuthToken, isTokenExpired, onAuthExpired, setAuthToken } from "./api"; import { + parseAdminChannelMessage, parseAgent, parseAwakenTool, parseGraph, @@ -85,6 +86,15 @@ function dispatch(ev: { type: string; data: unknown }): void { } break; } + case "channel_persisted": { + // design/16 §4.6 §4 — an isolated (group/stranger) channel message was + // persisted and broadcast for the IM adapter log page. The web chat has no + // handler for this type, so it never reaches the active conversation; the + // admin adapter log consumes it via `channelEvents` to append in realtime. + const msg = parseAdminChannelMessage(r?.message); + if (msg) s.pushChannelEvent(msg); + break; + } case "thinking.start": { pendingThinking = ""; break; diff --git a/frontend/web/src/lib/wsParse.ts b/frontend/web/src/lib/wsParse.ts index b797956..44802f9 100644 --- a/frontend/web/src/lib/wsParse.ts +++ b/frontend/web/src/lib/wsParse.ts @@ -1,4 +1,4 @@ -import { MODES, type AgentInfo, type ChatMessage, type FlowGraph, type Mode, type ToolUsed } from "./types"; +import { MODES, type AdminChannelMessage, type AgentInfo, type ChatMessage, type FlowGraph, type Mode, type ToolUsed } from "./types"; import { now, uid } from "./util"; /** @@ -158,3 +158,28 @@ export function parsePersistedMessage(v: unknown): ChatMessage | null { createdAt: typeof r.createdAt === "number" ? r.createdAt : now(), }; } + +/** Build an AdminChannelMessage from a `channel_persisted` payload (design/16 + * §4.6 §4): the backend broadcast a fully-persisted isolated (group/stranger) + * channel message carrying the backend id + adapter id, for the IM adapter log + * page to append in realtime. Requires id + sessionId; role/content fall back to + * defaults so a malformed frame is dropped rather than rendered. */ +export function parseAdminChannelMessage(v: unknown): AdminChannelMessage | null { + const r = toRecord(v); + if (!r) return null; + const id = typeof r.id === "string" && r.id ? r.id : ""; + const sessionId = typeof r.sessionId === "string" && r.sessionId ? r.sessionId : ""; + if (!id || !sessionId) return null; + return { + id, + sessionId, + role: typeof r.role === "string" ? r.role : "user", + content: typeof r.content === "string" ? r.content : "", + mode: typeof r.mode === "string" ? r.mode : undefined, + platform: typeof r.platform === "string" && r.platform ? r.platform : undefined, + deviceId: typeof r.deviceId === "string" && r.deviceId ? r.deviceId : undefined, + adapterId: typeof r.adapterId === "string" && r.adapterId ? r.adapterId : undefined, + senderName: typeof r.senderName === "string" && r.senderName ? r.senderName : undefined, + createdAt: typeof r.createdAt === "number" ? r.createdAt : now(), + }; +} diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index 72b761c..70e7c31 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -1,6 +1,7 @@ import { create } from "zustand"; import { persist } from "zustand/middleware"; import type { + AdminChannelMessage, AgentInfo, AwakenState, BackthinkItem, @@ -45,6 +46,13 @@ export interface AppStore { sessionBusy: boolean; pendingCount: number; + /** 适配器消息日志实时事件（`channel_persisted` WS 帧，design/16 §4.6 §4）—— + * 隔离会话（群聊/陌生人）落库后秒推，供 IM 适配器日志页增量插入。最新在前， + * 最多保留最近 ~200 条，按 id 去重。web 聊天无此字段的消费者。 */ + channelEvents: AdminChannelMessage[]; + /** Push one `channel_persisted` message (newest first, dedup by id, capped ~200). */ + pushChannelEvent: (msg: AdminChannelMessage) => void; + /** 当前 persona（web 主会话=默认 persona）。加载前/失败时为 null，头像回退中性默认占位。 */ currentPersona: PersonaRecord | null; @@ -116,6 +124,7 @@ export const useAppStore = create()( pendingCount: 0, currentPersona: null, taskLogs: {}, + channelEvents: [], setWsStatus: (status) => set({ wsStatus: status }), setCurrentPersona: (p) => set({ currentPersona: p }), @@ -316,6 +325,12 @@ export const useAppStore = create()( return { taskLogs }; }), + pushChannelEvent: (msg) => + set((s) => { + if (s.channelEvents.some((m) => m.id === msg.id)) return s; + return { channelEvents: [msg, ...s.channelEvents].slice(0, 200) }; + }), + resetChat: () => set((s) => ({ messages: [], diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 12c649d..7af2264 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -79,8 +79,9 @@ type SessionStore interface { // (design/16 §4.4): when a history store is wired, owner-private messages are // routed into the main web session and tagged with the channel platform, and every // persisted message is handed to OnPersisted so the gateway can broadcast a -// `message_persisted` event; when both are absent the bridge runs exactly as -// before (isolated sess-xxx, no chat_messages writes). +// `message_persisted` (web chat) or `channel_persisted` (IM adapter log) event; +// when both are absent the bridge runs exactly as before (isolated sess-xxx, no +// chat_messages writes). type Options struct { Log *slog.Logger Loop Chatter @@ -572,7 +573,8 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // Persist the inbound user message before the turn so it is durable even if the // channel reply fails. Owner-private messages land in the main web session and // are broadcast to the Web client; restricted messages land in their isolated - // session (the adapter log page) and are NOT broadcast to the main session. + // session (the adapter log page) and are broadcast as a `channel_persisted` event + // (never into the main/`web` session or the web chat). if isOwnerPrivate { b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg), adapterName(a)) } else { @@ -591,8 +593,9 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID) return } - // Persist the assistant reply (owner-private → main session + broadcast; - // restricted → isolated log) then deliver it back to the channel. + // Persist the assistant reply (owner-private → main session + message_persisted + // broadcast; restricted → isolated log + channel_persisted broadcast) then + // deliver it back to the channel. if isOwnerPrivate { b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg), adapterName(a)) } else { diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index 2b9b65b..5bfe513 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -377,7 +377,11 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) { chat := &stubChatter{res: &loop.Result{Text: "群聊回复"}} sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}} hist := &captureHistory{} - b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist}) + var persisted []history.Message + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + OnPersisted: func(m history.Message) { persisted = append(persisted, m) }, + }) if err != nil { t.Fatalf("New: %v", err) } @@ -411,6 +415,21 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) { if chat.sessionID != "sess-g" { t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID) } + // Each isolated write is ALSO handed to the broadcast hook so the gateway can + // emit `channel_persisted` for the IM adapter log page — but the message itself + // stays strictly in the isolated session and never leaks to the main/`web` + // session (design/16 §4.6 §3). + if len(persisted) != 2 { + t.Fatalf("isolated persisted notifications = %d, want 2", len(persisted)) + } + for i, p := range persisted { + if p.SessionID != "sess-g" { + t.Fatalf("persisted[%d] SessionID = %q, want isolated sess-g", i, p.SessionID) + } + if p.SessionID == history.MainSessionID { + t.Fatalf("persisted[%d] leaked to main web session", i) + } + } } func TestBlacklistGroupMessageDroppedBeforeLoop(t *testing.T) { diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index 0576cc7..a45e791 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -36,7 +36,8 @@ func (b *Bridge) SetOnRejected(fn func(adapter.UnifiedMessage, string)) { } // notifyPersisted hands a just-persisted message to the transport hook when one is -// wired, so the gateway can broadcast `message_persisted` for cross-end sync. +// wired, so the gateway can broadcast a `message_persisted` (web chat) or +// `channel_persisted` (IM adapter log) event for cross-end realtime sync. func (b *Bridge) notifyPersisted(msg history.Message) { b.mu.RLock() fn := b.onPersisted @@ -73,9 +74,11 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, dev // session (NOT the main web session) and carries the channel platform, the client // deviceID, the sender's display name and the account adapter registration name, so // the IM adapter log page can render the non-owner message stream per adapter config. -// It is deliberately NOT broadcast to the Web main session (only owner-private -// messages mirror to the Web). It is a no-op when no history store is wired, so the -// bridge runs unchanged without storage. +// The message is still NOT mirrored into the Web main session (only owner-private +// messages mirror to the Web); it is merely handed to the transport hook so the +// gateway broadcasts a `channel_persisted` event and the IM adapter log page appends +// it without a refresh. It is a no-op when no history store is wired, so the bridge +// runs unchanged without storage. func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID string, sess session.Session, platform, senderName, adapterID string) { if b.history == nil { return @@ -89,6 +92,10 @@ func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID st "session", sess.SessionID, "platform", platform, "sender_name", senderName, "adapter", adapterID) return } - // No broadcast: third-party messages must not reach the Web main session (design/16 §4.6 §3). - _ = full + // Broadcast the isolated write as a `channel_persisted` event so the IM adapter + // log page appends it in realtime. The gateway splits the event type by session: + // this is NOT the main web session, so the web chat has no handler for it and the + // message itself still lands only in the isolated sess-xxx session (never the + // main/`web` session, design/16 §4.6 §3). + b.notifyPersisted(full) } diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 95c9091..181b173 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -24,6 +24,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" @@ -935,6 +936,54 @@ func TestServerBroadcastCapacityAwaken(t *testing.T) { } } +// TestServerBroadcastPersistedSplit verifies BroadcastPersisted routes the event +// type by session: the main web session and its web sub-sessions broadcast +// `message_persisted` (web chat realtime sync), while an isolated channel adapter +// session (sess-xxx) broadcasts `channel_persisted` (IM adapter log page) so it is +// naturally ignored by the web chat handler. +func TestServerBroadcastPersistedSplit(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + c := newFakeConn(env.srv.hub, 32) + env.srv.hub.register(c) + + cases := []struct { + name string + sessionID string + wantType string + }{ + {name: "main web session", sessionID: history.MainSessionID, wantType: "message_persisted"}, + {name: "web sub session", sessionID: history.MainSessionID + ":abc123", wantType: "message_persisted"}, + {name: "isolated channel session", sessionID: "sess-g", wantType: "channel_persisted"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + env.srv.BroadcastPersisted(history.Message{ + ID: "m-" + tc.sessionID, SessionID: tc.sessionID, Role: history.RoleUser, + Content: "hello", Platform: "telegram", AdapterID: "telegram:abc", CreatedAt: 1000, + }) + ev := readFakeFrame(t, c) + if ev.Type != tc.wantType { + t.Fatalf("frame type = %q, want %q (frame=%+v)", ev.Type, tc.wantType, ev) + } + if ev.SessionID != tc.sessionID { + t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, tc.sessionID) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data = %+v, want map", ev.Data) + } + msg, ok := data["message"].(map[string]any) + if !ok { + t.Fatalf("frame data.message = %+v, want map", data["message"]) + } + if msg["content"] != "hello" || msg["adapterId"] != "telegram:abc" { + t.Fatalf("frame message = %+v, want content/hello + adapterId/telegram:abc", msg) + } + }) + } +} + func TestHubStreamChatIdle(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) // nil runner diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 4831f20..89a557e 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -25,6 +25,7 @@ import ( "net" "net/http" "strconv" + "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" @@ -514,17 +515,25 @@ func newOrchestrator(s *Server) *Orchestrator { }) } -// BroadcastPersisted broadcasts a just-persisted chat message as a -// `message_persisted` WS event so every browser of the same account appends it -// without a manual refresh (design/16 §4.2). It is shared by the orchestrator -// (Web turns) and the channel bridge (IM owner-private messages routed into the -// main web session), so both paths emit the identical frame. Thinking is stripped -// to stay consistent with the WS history replay (design/16 §6). It is a no-op-safe -// call when the hub has broadcast wired. +// BroadcastPersisted broadcasts a just-persisted chat message as a WS event so +// every browser of the same account appends it without a manual refresh +// (design/16 §4.2). It is shared by the orchestrator (Web turns) and the channel +// bridge (IM messages), so both paths flow through one frame shape. The event type +// is split by session: the main web session and its web sub-sessions broadcast +// `message_persisted` for web-chat realtime sync, while any other session id is an +// isolated channel adapter session (sess-xxx) and broadcasts `channel_persisted` +// for the IM adapter log page — the web chat has no handler for the latter, so it +// is naturally ignored there and never pollutes the conversation. Thinking is +// stripped to stay consistent with the WS history replay (design/16 §6). It is a +// no-op-safe call when the hub has broadcast wired. func (s *Server) BroadcastPersisted(msg history.Message) { wire := toHistoryMessage(msg) wire.Thinking = "" - s.hub.broadcast(wsEvent{Type: "message_persisted", SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) + typ := "channel_persisted" + if msg.SessionID == history.MainSessionID || strings.HasPrefix(msg.SessionID, history.MainSessionID+":") { + typ = "message_persisted" + } + s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) } // wsCapacityAwaken is the WS event type for a capacity state transition