Aska
6a2b8b9f12
backthink 决策+路由(P2)就绪但无投递 sink(ActiveClient/IMAdapter 无实现)。补齐: - gateway/proactive.go: hub.SendProactive 向 owner/admin 连接广播 proactive_message WS事件, 内容脱敏(去RawData/media/token),返回接收数,count=0→IM兜底;Server.SendProactive 委托 - channelbridge/proactive.go: Bridge.SendProactive 解析 owner 账户(msg.Platform→匹配白名单账号, 否则首个已配置主人频道),ChannelType=Direct+ChannelID/SenderID=owner id,走 adapter.SendProactive - backthink/sinks.go: SetActiveSink/SetIMSink 晚注入(New后重建pusher,currentPusher 并发安全) - wire.go: 在 gw/bridge 建好后 bt.SetActiveSink(gw)/SetIMSink(bridge) - 投递链: think→parseProactive→ShouldPush→currentPusher.Push→gate→active(active_client)/IM(im_bridge) →record(route+送达数) - 测试: sinks/gateway proactive/channelbridge proactive; 后端 build/vet/test 全绿 - 注: -race 本机 TSan 不可用(环境); 多平台IM按msg.Platform匹配 owner 账号(单sink诚实); weixin 24h窗口按 ErrProactiveWindow 记录为 im_bridge 错误不静默
41 lines
1.5 KiB
Go
41 lines
1.5 KiB
Go
package backthink
|
|
|
|
// The production composition root builds the gateway hub and the channel bridge
|
|
// *after* backthink.New (the hub needs the loop, the bridge needs the loop and
|
|
// owner profile, both of which are assembled later), so the ActiveClient and
|
|
// IMAdapter sinks are unavailable at construction time. These setters inject them
|
|
// once the gateway / bridge exist and rebuild the pusher, keeping the resolved
|
|
// rate gate so a later push routes correctly.
|
|
|
|
// SetActiveSink wires the production active web/native client sink (the gateway
|
|
// hub) and rebuilds the pusher. Call it after the gateway exists, before Start.
|
|
func (b *Backthink) SetActiveSink(active ActiveClient) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.deps.Active = active
|
|
b.rebuildPusherLocked()
|
|
}
|
|
|
|
// SetIMSink wires the production IM bridge sink (the channel bridge) and rebuilds
|
|
// the pusher. Call it after the channel bridge exists, before Start.
|
|
func (b *Backthink) SetIMSink(im IMAdapter) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.deps.IM = im
|
|
b.rebuildPusherLocked()
|
|
}
|
|
|
|
// rebuildPusherLocked replaces the pusher from the current sinks and gate. The
|
|
// caller must hold b.mu.
|
|
func (b *Backthink) rebuildPusherLocked() {
|
|
b.pusher = newPusher(b.deps.Active, b.deps.IM, b.gate, b.log)
|
|
}
|
|
|
|
// currentPusher returns the active pusher under the lifecycle lock, so a sink
|
|
// swap never races with a concurrent thinking pass that is about to push.
|
|
func (b *Backthink) currentPusher() Pusher {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return b.pusher
|
|
}
|