package channelbridge import ( "context" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" ) // 本文件承载 channelbridge 的聊天历史持久化钩子(design/16 // §4.4):主人的私聊被路由进唯一的 Web 主会话并 // 打上渠道平台标签,随后广播出去,使 Web 客户端无需刷新 // 即可镜像。它与 channelbridge.go 分开,好让桥接器的入站 // 流水线文件保持在模块体积上限之内。 // modeChat 是每条 IM 持久化消息写入的行为模式:IM 只承载 // 聊天文本(design/02 §1 单向回流),因此那里永远不会出现其他模式。 const modeChat = "chat" // SetOnPersisted 装配(或替换)主人私聊消息持久化进 Web 主会话 // 之后所调用的传输钩子。组合根中 gateway 在桥接器 // 之后构建,因此该钩子在 New 之后通过此 setter 挂上; // 它可以安全空调用,也可与在途处理 // 安全并发。 func (b *Bridge) SetOnPersisted(fn func(history.Message)) { b.mu.Lock() defer b.mu.Unlock() b.onPersisted = fn } // SetOnRejected 装配(或替换)消息被允许/拒绝黑白名单 // 丢弃后所调用的钩子(design/16 §4.6 §6)。组合根 // 把它接进 gateway 审计轨迹。它可以安全空调用,也可与 // 在途处理安全并发。 func (b *Bridge) SetOnRejected(fn func(adapter.UnifiedMessage, string)) { b.mu.Lock() defer b.mu.Unlock() b.onRejected = fn } // notifyPersisted 在装配了传输钩子时,把刚持久化的消息交给它, // 使 gateway 能广播 `message_persisted`(Web 聊天)或 // `channel_persisted`(IM 适配器日志)事件,实现多端实时同步。 func (b *Bridge) notifyPersisted(msg history.Message) { b.mu.RLock() fn := b.onPersisted b.mu.RUnlock() if fn != nil { fn(msg) } } // persistToMain 把一条消息写入 Web 主会话,并显式带上来源平台 // 与客户端 deviceID,随后通知传输钩子,让 Web 客户端无需刷新即可收到 // `message_persisted` 事件(design/16 §4.4)。deviceID 是 // 发送该消息的具体客户端(design/16 §4.5)——例如 IM 的 "weixin-ilink:conf0:wx_uid"。 // adapterID 是接收该消息的账号适配器注册名 // (如 "kook:3bef..."),便于管理日志按适配器查询收/发流; // Web 来源的消息该字段为空。mood 是该轮角色的心情 // (design/01 §2 情绪状态机),只挂在助手回复上(design/16 §3.3: // 助手消息承载 mode/mood);用户消息与未跟踪的心情传 ""。 // document 是这条消息携带的文档引用数组(design/24 §2.4、§6.1), // 形状即 history.EncodeDocuments 的产物;没有附件时传 ""(那一列因此为 // 空,读侧 `documents` 键随之省略)。IM 入站的附件与 Web 上传共用这一列, // 客户端两侧读到的附件芯片才是同一个东西。 // image 是这条消息携带的图片(data: URI 或 http(s) URL),写进 DTO 的 `image` // 字段(design/16 §4.4);没有图片时传 ""。它**不是** document 列的一部分: // 图片走视觉链路,文档走抽取,两者是两条路(design/24 §4.2 末的明确分野)。 // turnID 是这条消息所属的那**一路** IM 回合的权威标识(design/15 §2.5.9),由 // 驱动这次回合的桥接器生成(turnid.go):同一次入站回合的用户消息与助手回复传 // **同一个**值,前端才能按 (sessionId, turnId) 两级分桶。不属于任何回合的消息 // (主动推送等)传 "",那一列因此留空,行为与加入该列之前逐字相同。 // // 它返回刚落库的完整消息(未装配历史存储、或写库失败时是零值), // 使调用方能在回合结束后按 id 把文档抽取结论补进同一列 // (inbound_documents_persist.go 的两段式)。未装配历史存储时为空操作, // 因此桥接器在没有存储的情况下行为不变。 // // 写入本身走 persistMainWith(与主动推送同一个写入口径),通知走普通通道 notifyPersisted。 // **主动推送不经过本函数**:它直接调 persistMainWith,并额外走 notifyProactive 追加一条带 // 结构化判据的帧(切片 P0b)—— 因此带判据的帧只可能来自主动推送,**普通消息的帧永远 // 不带它**。 func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message { return b.persistMainWith(ctx, mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID), b.notifyPersisted) } // mainRow 构造一条写进 Web 主会话的行。它只写一次字段清单:persistToMain 与 // 入站语音那条「先写列再广播」的路径(coalesce_batch.go 的 persistBatchUsers) // 共用它,两处的列集合因此不可能分叉。 func mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message { return history.Message{ SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Document: document, Image: image, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat, TurnID: turnID, } } // persistMainWith 是「写进主会话」的**唯一**写入口径(与 persistIsolatedWith 同一手法): // persistToMain(普通消息)、主动推送(proactive.go)与入站/出站语音那条「先写 `voice` 列、 // 再广播」的路径共用它,因此几条路的存储语义逐字一致 —— 未接线历史存储时空操作并返回零值、 // 写库失败 WARN 并返回零值(绝不谎称已落库)、成功则返回刚落库的完整消息。 // // **差别只在 notify**:那是「这条要不要通知实时通道、以及先写列还是先广播」的接缝,由调用方 // 给出 —— 普通消息传 notifyPersisted(`message_persisted` / `channel_persisted`),主动推送 // 传 notifyProactive(带结构化判据的同一类帧,切片 P0b),语音传 persistVoiceFirstNotify // (先写 `voice` 列再广播,M44 首帧纪律)。nil 表示本次写入不通知。 func (b *Bridge) persistMainWith(ctx context.Context, msg history.Message, notify func(history.Message)) history.Message { if b.history == nil { return history.Message{} } full, err := b.history.AppendFull(ctx, msg) if err != nil { b.log.Warn("persist main session", "err", err, "role", msg.Role, "platform", msg.Platform, "device_id", msg.DeviceID, "adapter", msg.AdapterID) return history.Message{} } if notify != nil { notify(full) } return full } // persistIsolated 把一条消息写入受限(群聊/陌生人)会话 // 自有的平台日志(design/16 §4.6 §4):消息落在隔离的 sess-xxx // 会话(不是 Web 主会话)中,并带上渠道平台、客户端 // deviceID、发送者显示名与账号适配器注册名,便于 // IM 适配器日志页按适配器配置渲染非主人消息流。 // 该消息仍不会镜像进 Web 主会话(只有主人私聊 // 消息才镜像到 Web);它只交给传输钩子,让 // gateway 广播 `channel_persisted` 事件、IM 适配器日志页无需刷新 // 即可追加。document 是该消息携带的文档引用数组(design/24 §2.4), // 与 persistToMain 同一列、同一形状;没有附件时传 ""。image 是该消息携带的图片 // (data: URI 或 http(s) URL),同样与 persistToMain 同一列(DTO 的 `image` 字段); // 没有图片时传 ""。turnID 与 persistToMain 同义:这次入站回合的权威标识, // 不属于任何回合(被静默登记的消息)时传 ""。 // // 它返回刚落库的完整消息(未装配历史存储、或写库失败时是零值), // 供调用方在回合结束后补回文档抽取结论。未装配历史存储时为空操作, // 因此桥接器在没有存储的情况下行为不变。 func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID, turnID string) history.Message { return b.persistIsolatedWith(ctx, isolatedRow(role, content, document, image, deviceID, sess, platform, senderName, adapterID, turnID), b.notifyPersisted) } // isolatedRow 构造一条写进受限(群聊/陌生人)会话的行。与 mainRow 同一手法:字段清单 // 只写一次,persistIsolated 与入站语音那条「先写列再广播」的路径共用。 func isolatedRow(role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID, turnID string) history.Message { return history.Message{ SessionID: sess.SessionID, Role: role, Content: content, Document: document, Image: image, Platform: platform, DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat, TurnID: turnID, } } // persistIsolatedWith 是 persistIsolated 的**可注入通知**变体:写入语义逐字相同,只有 // 「落库成功后怎么通知」由调用方给出(nil = 不通知)。入站语音用它把「先写 `voice` 列、 // 再广播」这次换序塞进同一个落库点(见 inbound_voice_copy.go 的 persistVoiceFirstNotify)。 func (b *Bridge) persistIsolatedWith(ctx context.Context, row history.Message, notify func(history.Message)) history.Message { if b.history == nil { return history.Message{} } full, err := b.history.AppendFull(ctx, row) if err != nil { b.log.Warn("persist isolated session", "err", err, "role", row.Role, "session", row.SessionID, "platform", row.Platform, "sender_name", row.SenderName, "adapter", row.AdapterID) return history.Message{} } // 把隔离写入作为 `channel_persisted` 事件广播,让 IM 适配器 // 日志页实时追加。gateway 按会话区分事件类型: // 这不是 Web 主会话,因此 Web 聊天没有它的处理器, // 消息本身仍只落在隔离的 sess-xxx 会话(绝不在 // 主/`web` 会话,design/16 §4.6 §3)。 if notify != nil { notify(full) } return full }