From 6a2b8b9f12c1d534507ff43e6b8160dfcee8dc1f Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 19:43:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(backthink):=20=E4=B8=BB=E5=8A=A8=E6=8E=A8?= =?UTF-8?q?=E9=80=81=20sink=20=E6=89=93=E9=80=9A=E2=80=94=E2=80=94hub.Send?= =?UTF-8?q?Proactive=20+=20bridge.SendProactive=20+=20wire=20=E6=8E=A5?= =?UTF-8?q?=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 错误不静默 --- cmd/cyrene/wire.go | 12 ++ internal/services/backthink/backthink.go | 13 +- internal/services/backthink/sinks.go | 40 +++++ internal/services/backthink/sinks_test.go | 102 +++++++++++++ .../channelbridge/channelbridge_test.go | 5 +- internal/services/channelbridge/proactive.go | 100 +++++++++++++ .../services/channelbridge/proactive_test.go | 137 ++++++++++++++++++ internal/services/gateway/proactive.go | 89 ++++++++++++ internal/services/gateway/proactive_test.go | 123 ++++++++++++++++ 9 files changed, 615 insertions(+), 6 deletions(-) create mode 100644 internal/services/backthink/sinks.go create mode 100644 internal/services/backthink/sinks_test.go create mode 100644 internal/services/channelbridge/proactive.go create mode 100644 internal/services/channelbridge/proactive_test.go create mode 100644 internal/services/gateway/proactive.go create mode 100644 internal/services/gateway/proactive_test.go diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 347fd33..202369e 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -662,6 +662,18 @@ func (a *app) wireFrontend() error { } a.gw = gw a.capacitySink = gw + + // Wire the backthink proactive sinks now that the gateway hub (active-client + // route) and the channel bridge (IM fallback) both exist. They were absent when + // backthink.New ran (the hub needs the loop and the bridge needs the loop and the + // owner profile, all assembled later), so without this the pusher would resolve + // RouteNone. Each setter rebuilds the pusher under the service's lifecycle lock. + if a.backthink != nil { + a.backthink.SetActiveSink(gw) + if a.channelBridge != nil { + a.backthink.SetIMSink(a.channelBridge) + } + } return nil } diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 046691f..705018b 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -157,6 +157,7 @@ type Backthink struct { log *slog.Logger store *Store pusher Pusher + gate PushGate now func() time.Time sessionID string @@ -187,12 +188,13 @@ func New(cfg Config, deps Deps) (*Backthink, error) { return nil, fmt.Errorf("backthink: new: %w", err) } + // Resolve the gate up front so the sink setters can rebuild the pusher with it. + gate := deps.Gate + if gate == nil { + gate = newRateGate(cfg.MaxPushWindow, cfg.PushWindow, log) + } pusher := deps.Pusher if pusher == nil { - gate := deps.Gate - if gate == nil { - gate = newRateGate(cfg.MaxPushWindow, cfg.PushWindow, log) - } pusher = newPusher(deps.Active, deps.IM, gate, log) } @@ -211,6 +213,7 @@ func New(cfg Config, deps Deps) (*Backthink, error) { log: log, store: store, pusher: pusher, + gate: gate, now: now, sessionID: sessionID, }, nil @@ -407,7 +410,7 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { pushRes := PushResult{Route: RouteNone, Detail: b.interruptReason(push, at, state)} if shouldPush { - pushRes = b.pusher.Push(cctx, PushRequest{ + pushRes = b.currentPusher().Push(cctx, PushRequest{ Message: b.buildProactiveMessage(message, channelHint), ChannelHint: channelHint, }) diff --git a/internal/services/backthink/sinks.go b/internal/services/backthink/sinks.go new file mode 100644 index 0000000..bbd8cd8 --- /dev/null +++ b/internal/services/backthink/sinks.go @@ -0,0 +1,40 @@ +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 +} diff --git a/internal/services/backthink/sinks_test.go b/internal/services/backthink/sinks_test.go new file mode 100644 index 0000000..92a12f5 --- /dev/null +++ b/internal/services/backthink/sinks_test.go @@ -0,0 +1,102 @@ +package backthink + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// TestBackthink_SetActiveSink_RebuildsPusher verifies the sink setters rebuild the +// pusher so a later push routes to the injected active client instead of the +// default nil-sink RouteNone pusher built at New. +func TestBackthink_SetActiveSink_RebuildsPusher(t *testing.T) { + db := newTestDB(t) + bt, err := New(Config{}, Deps{ + Thinker: &mockThinker{text: "[我决定:push]\n[渠道:web]\nhi"}, + State: &mockState{}, + Gate: &mockGate{allow: true}, + DB: db, + }) + if err != nil { + t.Fatalf("new: %v", err) + } + + active := &mockActive{n: 2} + im := &mockIMAdapter{} + bt.SetActiveSink(active) + bt.SetIMSink(im) + + res := bt.pusher.Push(context.Background(), PushRequest{ + Message: adapter.UnifiedMessage{Content: "hi"}, + ChannelHint: "web", + }) + if !res.Pushed || res.Route != RouteActiveClient { + t.Fatalf("push = %+v, want active_client pushed", res) + } + if active.calls != 1 { + t.Fatalf("active calls = %d, want 1", active.calls) + } + if im.count() != 0 { + t.Fatalf("im calls = %d, want 0 (active preferred)", im.count()) + } +} + +// TestBackthink_SetIMSink_Fallback verifies a wired IM sink receives the push when +// no active client is online (so the owner is still reached). +func TestBackthink_SetIMSink_Fallback(t *testing.T) { + db := newTestDB(t) + bt, err := New(Config{}, Deps{ + Thinker: &mockThinker{text: "[我决定:push]\nhi"}, + State: &mockState{}, + Gate: &mockGate{allow: true}, + DB: db, + }) + if err != nil { + t.Fatalf("new: %v", err) + } + + active := &mockActive{n: 0} // no active client online + im := &mockIMAdapter{} + bt.SetActiveSink(active) + bt.SetIMSink(im) + + res := bt.pusher.Push(context.Background(), PushRequest{ + Message: adapter.UnifiedMessage{Content: "hi"}, + }) + if !res.Pushed || res.Route != RouteIMBridge { + t.Fatalf("push = %+v, want im_bridge pushed", res) + } + if im.count() != 1 { + t.Fatalf("im calls = %d, want 1", im.count()) + } +} + +// TestBackthink_SinkSetters_ConcurrentSafe verifies the setter and the read of the +// current pusher do not race under the lifecycle lock, so an in-flight pass never +// touches a half-rebuilt pusher. +func TestBackthink_SinkSetters_ConcurrentSafe(t *testing.T) { + db := newTestDB(t) + bt, err := New(Config{}, Deps{ + Thinker: &mockThinker{}, + State: &mockState{}, + Gate: &mockGate{allow: true}, + DB: db, + }) + if err != nil { + t.Fatalf("new: %v", err) + } + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 200; i++ { + bt.SetActiveSink(&mockActive{n: 1}) + bt.SetIMSink(&mockIMAdapter{}) + } + }() + for i := 0; i < 200; i++ { + _ = bt.currentPusher().Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{Content: "x"}}) + } + <-done +} diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index 5bfe513..c239c60 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -50,7 +50,10 @@ func (f *fakeAdapter) SendMessage(_ context.Context, resp adapter.UnifiedRespons } return f.sendErr } -func (f *fakeAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { return nil } +func (f *fakeAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error { + f.sends = append(f.sends, msg) + return f.sendErr +} func (f *fakeAdapter) SetMessageHandler(fn adapter.MessageHandler) { f.handler = fn } func (f *fakeAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() } func (f *fakeAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { return nil, nil } diff --git a/internal/services/channelbridge/proactive.go b/internal/services/channelbridge/proactive.go new file mode 100644 index 0000000..953d8ba --- /dev/null +++ b/internal/services/channelbridge/proactive.go @@ -0,0 +1,100 @@ +package channelbridge + +import ( + "context" + "fmt" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// proactiveTarget is the owner's IM address (platform + account id) resolved for +// a proactive push. The account id doubles as the DM peer on every adapter. +type proactiveTarget struct { + platform string + accountID string +} + +// SendProactive pushes msg to the owner's IM channel via the registered adapter +// for that platform. It is the backthink IMAdapter seam (design/12 §3 fallback). +// The target account is chosen by msg.Platform when it names one of the owner's +// whitelisted channels, otherwise the first whitelisted owner channel is used. +// When the owner profile has no whitelisted account (or none carries an account +// id) it returns a clear error so the caller records the reason rather than +// silently dropping the push. +func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { + if err := ctx.Err(); err != nil { + return err + } + target, ok := b.proactiveTarget(msg) + if !ok { + return fmt.Errorf("channelbridge: no owner IM channel for proactive push") + } + ad := b.adapterForPlatform(target.platform) + if ad == nil { + return fmt.Errorf("channelbridge: no adapter for owner platform %q", target.platform) + } + out := msg + out.Platform = target.platform + out.ChannelType = adapter.ChannelDirect + // The owner's account id is the DM peer on every adapter: telegram/qq read + // ChannelID, kook/weixin read SenderID (with ChannelID as the fallback). Setting + // both keeps a direct message targetable regardless of which field the adapter + // prefers, without duplicating the message. + out.ChannelID = target.accountID + out.SenderID = target.accountID + if err := ad.SendProactive(ctx, out); err != nil { + return fmt.Errorf("channelbridge: send proactive via %q: %w", target.platform, err) + } + return nil +} + +// proactiveTarget resolves the owner account a proactive push should reach: +// msg.Platform, when it names one of the owner's whitelisted platforms, selects +// that account; otherwise the first whitelisted owner channel (with a non-empty +// account id) is used. A nil owner or an empty whitelist yields !ok. +func (b *Bridge) proactiveTarget(msg adapter.UnifiedMessage) (proactiveTarget, bool) { + if b.owner == nil { + return proactiveTarget{}, false + } + accounts := b.owner.Accounts + if len(accounts) == 0 { + return proactiveTarget{}, false + } + platform := strings.ToLower(strings.TrimSpace(msg.Platform)) + if platform != "" { + for _, acct := range accounts { + if acct.AccountID != "" && strings.EqualFold(acct.Platform, platform) { + return proactiveTarget{acct.Platform, acct.AccountID}, true + } + } + } + for _, acct := range accounts { + if acct.AccountID != "" { + return proactiveTarget{acct.Platform, acct.AccountID}, true + } + } + return proactiveTarget{}, false +} + +// adapterForPlatform returns a registered adapter whose base platform matches +// platform (multi-account names like "kook:3be..." resolve to their bare base), +// preferring a connected one so a dead account is only tried when it is the only +// option. It returns nil when none is registered. +func (b *Bridge) adapterForPlatform(platform string) adapter.PlatformAdapter { + b.mu.RLock() + defer b.mu.RUnlock() + var first adapter.PlatformAdapter + for name, a := range b.adapters { + if !strings.EqualFold(basePlatform(name), platform) { + continue + } + if a.IsConnected() { + return a + } + if first == nil { + first = a + } + } + return first +} diff --git a/internal/services/channelbridge/proactive_test.go b/internal/services/channelbridge/proactive_test.go new file mode 100644 index 0000000..315715c --- /dev/null +++ b/internal/services/channelbridge/proactive_test.go @@ -0,0 +1,137 @@ +package channelbridge + +import ( + "context" + "errors" + "strings" + "testing" + + "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/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +func newProactiveBridge(t *testing.T, owner *user.OwnerProfile, ad *fakeAdapter) *Bridge { + t.Helper() + chat := &stubChatter{res: &loop.Result{Text: "hi"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + b, err := New(Options{Loop: chat, Sessions: sess, Owner: owner}) + if err != nil { + t.Fatalf("new bridge: %v", err) + } + if ad != nil { + b.Add(ad) + } + return b +} + +// proactiveMsg builds a minimal autonomous proactive message as backthink's +// buildProactiveMessage would (no channel targeting yet). +func proactiveMsg(content string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "cyrene-backthink", + ChannelType: adapter.ChannelDirect, + Content: content, + ContentType: adapter.ContentText, + Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"}, + } +} + +// TestBridge_SendProactive_OwnerChannel verifies a proactive push lands on the +// owner's IM channel: the adapter for the owner's platform is selected and the +// owner account id is threaded as the DM peer. +func TestBridge_SendProactive_OwnerChannel(t *testing.T) { + ad := &fakeAdapter{name: "telegram", connected: true} + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "telegram", AccountID: "100"}}} + b := newProactiveBridge(t, owner, ad) + + if err := b.SendProactive(context.Background(), proactiveMsg("记得喝水哦")); err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if len(ad.sends) != 1 { + t.Fatalf("sends = %d, want 1", len(ad.sends)) + } + got := ad.sends[0] + if got.Platform != "telegram" { + t.Fatalf("platform = %q, want telegram (owner account platform)", got.Platform) + } + if got.Content != "记得喝水哦" { + t.Fatalf("content = %q, want the proactive text", got.Content) + } + if got.ChannelID != "100" || got.SenderID != "100" { + t.Fatalf("peer = %q/%q, want the owner account id 100/100", got.ChannelID, got.SenderID) + } + if got.ChannelType != adapter.ChannelDirect { + t.Fatalf("channel type = %q, want direct", got.ChannelType) + } +} + +// TestBridge_SendProactive_PlatformHintSelectsAccount verifies msg.Platform +// selects the matching owner account when one exists, rather than defaulting. +func TestBridge_SendProactive_PlatformHintSelectsAccount(t *testing.T) { + ad := &fakeAdapter{name: "kook", connected: true} + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{ + {Platform: "telegram", AccountID: "100"}, + {Platform: "kook", AccountID: "329284282"}, + }} + b := newProactiveBridge(t, owner, ad) + + msg := proactiveMsg("洗完了") + msg.Platform = "kook" + if err := b.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if len(ad.sends) != 1 { + t.Fatalf("sends = %d, want 1", len(ad.sends)) + } + got := ad.sends[0] + if got.Platform != "kook" || got.SenderID != "329284282" { + t.Fatalf("target = %q/%q, want the kook owner account", got.Platform, got.SenderID) + } +} + +// TestBridge_SendProactive_NoOwnerChannel verifies a clearly-worded error when the +// owner profile has no whitelisted account, so the caller records the reason. +func TestBridge_SendProactive_NoOwnerChannel(t *testing.T) { + b := newProactiveBridge(t, &user.OwnerProfile{}, nil) + + err := b.SendProactive(context.Background(), proactiveMsg("hi")) + if err == nil { + t.Fatalf("expected an error for a missing owner channel, got nil") + } + if !strings.Contains(err.Error(), "no owner IM channel") { + t.Fatalf("error = %q, want a 'no owner IM channel' note", err) + } +} + +// TestBridge_SendProactive_NoAdapter verifies the error names the platform when the +// owner has a whitelisted channel but no adapter is registered for it. +func TestBridge_SendProactive_NoAdapter(t *testing.T) { + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "kook", AccountID: "329284282"}}} + b := newProactiveBridge(t, owner, nil) + + err := b.SendProactive(context.Background(), proactiveMsg("hi")) + if err == nil { + t.Fatalf("expected an error for a missing adapter, got nil") + } + if !strings.Contains(err.Error(), "no adapter for owner platform") { + t.Fatalf("error = %q, want a 'no adapter for owner platform' note", err) + } +} + +// TestBridge_SendProactive_AdapterError verifies an adapter send failure surfaces +// as a wrapped error with the platform (so the record/detail is honest). +func TestBridge_SendProactive_AdapterError(t *testing.T) { + ad := &fakeAdapter{name: "kook", connected: true, sendErr: errors.New("kook disconnected")} + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "kook", AccountID: "329284282"}}} + b := newProactiveBridge(t, owner, ad) + + err := b.SendProactive(context.Background(), proactiveMsg("hi")) + if err == nil { + t.Fatalf("expected an error, got nil") + } + if !strings.Contains(err.Error(), "kook") || !strings.Contains(err.Error(), "kook disconnected") { + t.Fatalf("error = %q, want it to wrap the adapter error and name the platform", err) + } +} diff --git a/internal/services/gateway/proactive.go b/internal/services/gateway/proactive.go new file mode 100644 index 0000000..e9e506c --- /dev/null +++ b/internal/services/gateway/proactive.go @@ -0,0 +1,89 @@ +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// wsProactiveMessage is the WS event type for an autonomous proactive push +// (design/12 §3): a message Cyrene decided to send without a recent user input. +const wsProactiveMessage = "proactive_message" + +// proactivePayload is the sanitized wire shape of a proactive message. It carries +// only the fields a client needs to render a bubble (content, platform, channel, +// sender label); it deliberately omits RawData / media bytes / tokens so a +// proactive push can never leak a credential or payload into the frontend. +type proactivePayload struct { + Platform string `json:"platform,omitempty"` + ChannelID string `json:"channelId,omitempty"` + ChannelType string `json:"channelType,omitempty"` + Content string `json:"content"` + ContentType string `json:"contentType,omitempty"` + Sender string `json:"sender,omitempty"` + Timestamp time.Time `json:"timestamp,omitempty"` + MessageID string `json:"messageId,omitempty"` +} + +// SendProactive delivers msg to every live owner/admin connection and returns how +// many accepted it. A zero count means no owner/admin client is online, so the +// caller falls back to the IM bridge. Only owner/admin connections are targeted +// (red line: proactive is owner-only); a viewer or unauthenticated connection is +// never reached. It never blocks: a full send buffer drops the frame for that +// slow client rather than stalling the backthink push. +func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg)}) + if err != nil { + return 0, fmt.Errorf("gateway: marshal proactive: %w", err) + } + h.mu.RLock() + defer h.mu.RUnlock() + n := 0 + for c := range h.conns { + if !ownerAdminClaims(c.claims) { + continue + } + select { + case c.send <- b: + n++ + default: + } + } + return n, nil +} + +// Server.SendProactive satisfies the backthink ActiveClient seam by delegating to +// the hub, so the composition root can hand the gateway itself to the sink setter. +func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) { + return s.hub.SendProactive(ctx, msg) +} + +// proactivePayloadFrom projects a UnifiedMessage into the sanitized wire payload. +func proactivePayloadFrom(msg adapter.UnifiedMessage) proactivePayload { + return proactivePayload{ + Platform: msg.Platform, + ChannelID: msg.ChannelID, + ChannelType: string(msg.ChannelType), + Content: msg.Content, + ContentType: string(msg.ContentType), + Sender: msg.Sender.DisplayName, + Timestamp: msg.Timestamp, + MessageID: msg.MessageID, + } +} + +// ownerAdminClaims reports whether claims authorize a proactive push (owner or +// admin). Nil claims (unauth / bare test connections) are never targeted. +func ownerAdminClaims(claims *auth.Claims) bool { + if claims == nil { + return false + } + return claims.Role == auth.RoleOwner || claims.Role == auth.RoleAdmin +} diff --git a/internal/services/gateway/proactive_test.go b/internal/services/gateway/proactive_test.go new file mode 100644 index 0000000..00c9f25 --- /dev/null +++ b/internal/services/gateway/proactive_test.go @@ -0,0 +1,123 @@ +package gateway + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// TestHub_SendProactive_OwnerAndAdmin verifies a proactive push reaches exactly +// the owner/admin connections and returns the delivered count; a viewer or +// unauthenticated (nil-claims) connection never receives it. +func TestHub_SendProactive_OwnerAndAdmin(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + owner := newFakeConn(hb, 8) + owner.claims = &auth.Claims{Role: auth.RoleOwner} + admin := newFakeConn(hb, 8) + admin.claims = &auth.Claims{Role: auth.RoleAdmin} + viewer := newFakeConn(hb, 8) + viewer.claims = &auth.Claims{Role: auth.RoleViewer} + bare := newFakeConn(hb, 8) // claims nil: unauthenticated + for _, c := range []*wsConn{owner, admin, viewer, bare} { + hb.register(c) + } + + msg := adapter.UnifiedMessage{ + Platform: "cyrene-backthink", + ChannelType: adapter.ChannelDirect, + Content: "记得喝水哦", + ContentType: adapter.ContentText, + Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"}, + ChannelID: "kook:123", + SenderID: "uid-999", + MessageID: "m-42", + } + n, err := hb.SendProactive(context.Background(), msg) + if err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if n != 2 { + t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n) + } + + for _, c := range []*wsConn{owner, admin} { + ev := readFakeFrame(t, c) + if ev.Type != "proactive_message" { + t.Fatalf("frame type = %q, want proactive_message (frame=%+v)", ev.Type, ev) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data type = %T, want map[string]any", ev.Data) + } + if data["content"] != "记得喝水哦" { + t.Fatalf("frame content = %q, want the proactive text", data["content"]) + } + if data["sender"] != "Cyrene" { + t.Fatalf("frame sender = %q, want Cyrene", data["sender"]) + } + if data["platform"] != "cyrene-backthink" { + t.Fatalf("frame platform = %q, want cyrene-backthink", data["platform"]) + } + } + for _, c := range []*wsConn{viewer, bare} { + if err := connHasNoFrame(c); err != nil { + t.Fatalf("unexpected frame for a non-owner/admin conn: %v", err) + } + } +} + +// TestHub_SendProactive_NoOnlineOwner verifies a zero count (no error) when no +// owner/admin connection is live, so the caller falls back to the IM bridge. +func TestHub_SendProactive_NoOnlineOwner(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + viewer := newFakeConn(hb, 8) + viewer.claims = &auth.Claims{Role: auth.RoleViewer} + hb.register(viewer) + + n, err := hb.SendProactive(context.Background(), adapter.UnifiedMessage{Content: "hi"}) + if err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if n != 0 { + t.Fatalf("count = %d, want 0", n) + } +} + +// TestHub_SendProactive_CancelledContext verifies a cancelled context denies the +// push (fail closed) with a wrapped error. +func TestHub_SendProactive_CancelledContext(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := hb.SendProactive(ctx, adapter.UnifiedMessage{Content: "hi"}); err == nil { + t.Fatalf("expected a context error, got nil") + } +} + +// TestServer_SendProactive_DelegatesToHub verifies Server.SendProactive mirrors the +// hub so the composition root can hand the gateway itself to the sink setter. +func TestServer_SendProactive_DelegatesToHub(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + c := newFakeConn(env.srv.hub, 8) + c.claims = &auth.Claims{Role: auth.RoleOwner} + env.srv.hub.register(c) + + n, err := env.srv.SendProactive(context.Background(), adapter.UnifiedMessage{Content: "hi"}) + if err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if n != 1 { + t.Fatalf("count = %d, want 1", n) + } + ev := readFakeFrame(t, c) + if ev.Type != "proactive_message" { + t.Fatalf("frame type = %q, want proactive_message", ev.Type) + } +}