Cyrene_v2/internal/services/backthink/push.go

package backthink

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
)

// Route 标识一条主动消息实际是如何投递的，供监控面板
// 与推送结果记录使用。
type Route string

const (
	// RouteNone 表示未尝试推送，或无可用目标。
	RouteNone Route = "none"
	// RouteRateLimited 表示推送被限流器拒绝。
	RouteRateLimited Route = "rate_limited"
	// RouteActiveClient 表示消息发往了活跃 Web/原生客户端。
	RouteActiveClient Route = "active_client"
	// RouteIMBridge 表示消息降级到 IM 桥。
	RouteIMBridge Route = "im_bridge"
)

// 推送路由器识别的渠道提示词。它们来自结论的
// [渠道:...] 标记（parseProactive），用于选择主动消息
// 的投递去向。为空/未知的提示按 auto 处理。具体的
// 平台 id（kook/telegram/...）是这三个之外的值。
const (
	channelAuto = "auto"
	channelIM   = "im"
	channelWeb  = "web"
	// defaultProactivePlatform 是结论未指明具体渠道时，
	// 出站主动消息携带的 Platform。
	defaultProactivePlatform = "cyrene-backthink"
)

// String 把路由作为普通标签返回。
func (r Route) String() string { return string(r) }

// PushResult 是一次主动推送尝试的结果，随每轮思考
// 一起记录（design/12 §4）。
type PushResult struct {
	// Pushed 报告消息是否已投递。
	Pushed bool
	// Route 是实际采用的路由。
	Route Route
	// Bypassed 报告本次推送是否为事件驱动、跳过了 PushGate
	// （真正的事件，"该发就发"）。心跳推送受门控管辖，绝不
	// 被绕过。
	Bypassed bool
	// Detail 是简短可读的备注（如客户端数量、决策理由）。
	Detail string
	// Err 是投递失败时的底层错误文本，或决定不推送的
	// 提供者的错误文本。
	Err string
}

// String 返回用于日志的简洁摘要。
func (p PushResult) String() string {
	if p.Err != "" {
		return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q err=%q", p.Pushed, p.Route, p.Bypassed, p.Detail, p.Err)
	}
	return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q", p.Pushed, p.Route, p.Bypassed, p.Detail)
}

// ActiveClient 把主动消息投递给所有当前活跃的 Web/原生
// 客户端，并报告有多少个收到。它是首选的推送目标。
// 生产中由 gateway 的 WebSocket hub 实现；测试注入 mock。
type ActiveClient interface {
	// SendProactive 把 msg 推给每个活跃客户端，并返回
	// 接受它的客户端数量。
	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error)
}

// IMAdapter 是 IM 桥降级：它把主动消息经 IM
// 渠道推送。每个 adapter.PlatformAdapter 都满足这个子集，因此本服务
// 依赖这个窄接口而非具体适配器类型。
type IMAdapter interface {
	// SendProactive 把 msg 推给主人的 IM 渠道。
	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error
}

// PushGate 判断一次主动推送此刻能否进行，套用
// design/11 §6 的限频规则，外加按风格限制与推送后冷却。它
// 是对 channel/session 的 ProactivePolicy/RateLimiter 的接缝，因此本服务
// 可以用 mock 测试。
type PushGate interface {
	// Allow 报告给定风格的主动推送此刻能否发送。
	Allow(ctx context.Context, kind PushKind) bool
	// RecordPush 记下一次成功投递，使冷却/按风格窗口
	// 从一次真正的成功推送起算，而不是仅仅尝试过。
	RecordPush(kind PushKind)
}

// PushRequest 携带一次主动推送尝试：出站消息加上
// 自主结论所要求的目标渠道（design/20 §2 P2）。
// ChannelHint 是已转小写的 [渠道:...] 词：""/auto 选默认（
// 主人要求：只发一个渠道，由她 / 她最近的平台 /
// 白名单默认决定）；"web" 指向活跃客户端；"im" 指向 IM 桥；
// 具体平台 id（kook/telegram/weixin-ilink/...）指向 IM 桥
// 并把消息标记为该平台。
type PushRequest struct {
	// Message 是归一化后的出站主动消息。
	Message adapter.UnifiedMessage
	// ChannelHint 是请求的目标渠道，她未选择时为 ""。
	ChannelHint string
	// TriggerKind 记录唤醒本轮的原因。事件驱动的推送绕过
	// PushGate（主人要求："该发就发"）；心跳保持受门控管辖。
	TriggerKind TriggerKind
}

// Pusher 把主动消息投递到一个目标渠道。它受
// 频率门约束，并尊重自主结论所选的渠道提示："web"
// 去活跃客户端，其余都经 IM 桥走一次，
// 由桥挑出那一个渠道（她的选择 → 她最近的平台 →
// 白名单默认，微信含在内）。
type Pusher interface {
	// Push 按渠道提示把 req 投递到单个目标渠道。它绝不
	// 阻塞在慢目标上。
	Push(ctx context.Context, req PushRequest) PushResult
}

// pusher 是具体的 Pusher：它先咨询频率门，再把消息路由到
// 恰好一个目标——"web" 提示走活跃客户端，否则走 IM 桥。
// 本地投递之后，它把消息镜像到可选的推送出口
// （自托管服务 → 个推降级），且不阻塞该投递。
type pusher struct {
	active   ActiveClient
	im       IMAdapter
	gate     PushGate
	exporter PushExporter
	log      *slog.Logger
	// exportDone 非 nil 时，会在每次异步导出完成后收到一个值。
	// 它存在是为了让测试能确定性地观测该钩子；发送是非阻塞的，
	// 所以通道满也不会卡住导出器。
	exportDone chan struct{}
}

// newPusher 由可选的活跃客户端槽、IM 槽与
// 频率门构建 Pusher。它们都可以为 nil；目标为 nil 只是让那条路由
// 不可用。推送出口用 withExporter 单独挂上。
func newPusher(active ActiveClient, im IMAdapter, gate PushGate, log *slog.Logger) *pusher {
	return &pusher{active: active, im: im, gate: gate, log: log}
}

// withExporter 挂上推送出口钩子并返回 pusher 以便链式
// 调用。exporter 为 nil 则钩子保持关闭。
func (p *pusher) withExporter(exporter PushExporter) *pusher {
	p.exporter = exporter
	return p
}

// Push 实现 Pusher：它套用频率门，然后把消息路由到
// 单个目标。"web" 提示到达活跃 Web/原生客户端；其它任何
// 提示（"im"、"auto" 或具体平台）经 IM 桥走一次，由
// 桥挑出一个主人渠道（她的 [渠道:...] 选择，其次是她
// 最近用过的平台，再次是白名单默认）。IM 桥是主路径，所以一次
// 推送绝不扇出到多个渠道。未接线 IM 槽时，活跃
// 客户端就是诚实的降级，因此主人仍能被触达，而不是
// 消息被丢弃。
func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult {
	if err := ctx.Err(); err != nil {
		return PushResult{Pushed: false, Route: RouteNone, Err: err.Error()}
	}
	kind := classifyPush(req.Message.Content)
	// 真正事件驱动的推送绕过 PushGate：它应原样发出（"该发就发"），
	// 不受心跳限流约束（60min≤1 / 冷却 / 按风格窗口）。
	// 只有周期性心跳保持受门控管辖，主人因此绝不会被
	// 空洞的反思刷屏。同一事件去重在 think() 上游套用。
	eventDriven := req.TriggerKind == TriggerEvent
	if p.gate != nil && !eventDriven && !p.gate.Allow(ctx, kind) {
		p.log.Warn("backthink push rate limited", "kind", kind.String())
		return PushResult{Pushed: false, Route: RouteRateLimited, Detail: "proactive push rate limited"}
	}

	hint := normalizeChannelHint(req.ChannelHint)
	msg := req.Message

	var res PushResult
	if hint == channelWeb {
		res = p.pushActive(ctx, msg)
	} else {
		// "im"/"auto"/具体平台都经 IM 桥走一次。
		// 泛化提示交出哨兵平台，让桥落到它的
		// 最近活跃 / 默认选择上；具体提示在渠道
		// 已加入白名单且已连接时被尊重。
		if hint == channelIM || hint == channelAuto {
			msg.Platform = defaultProactivePlatform
		} else {
			msg.Platform = hint
		}
		res = p.pushIM(ctx, msg)
		if !res.Pushed {
			res = p.pushActive(ctx, msg)
		}
	}
	// 只有真正的投递才推进冷却/按风格窗口（且只对
	// 受门控的推送），因此尝试过但失败的推送不会抑制之后的
	// 真实推送，而事件推送也绝不计入心跳限流。
	if res.Pushed && p.gate != nil && !eventDriven {
		p.gate.RecordPush(kind)
	}
	// D1：把主动消息镜像到外部推送出口（自托管
	// 统一推送服务 → 个推降级）。发后不管：慢或失败的
	// 出口绝不延迟或破坏上面的本地投递，而且限流被拒的
	// 推送在此之前就已返回。
	p.exportAsync(msg)
	res = annotateHint(res, req.ChannelHint)
	if eventDriven {
		res.Bypassed = true
		res = annotateBypass(res)
	}
	return res
}

// exportAsync 在脱离的、有界的 goroutine 上把 msg 上报给推送出口，
// 因此主动路径绝不在出口上阻塞。exporter 为 nil（钩子关闭）
// 或出口结果为 nil 都是空操作；投递未命中或出错只记日志。
func (p *pusher) exportAsync(msg adapter.UnifiedMessage) {
	if p.exporter == nil {
		return
	}
	timeout := defaultExportTimeout
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		defer cancel()
		delivered, err := p.exporter.Export(ctx, msg)
		switch {
		case err != nil:
			p.log.Warn("push outlet export failed", "error", err)
		case !delivered:
			p.log.Info("push outlet did not deliver", "hint", "getui fallback not integrated")
		}
		if p.exportDone != nil {
			select {
			case p.exportDone <- struct{}{}:
			default:
			}
		}
	}()
}

// pushActive 只投递给活跃 Web/原生客户端。
func (p *pusher) pushActive(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
	if p.active != nil {
		n, err := p.active.SendProactive(ctx, msg)
		if err != nil {
			p.log.Warn("backthink active client push failed", "error", err)
		} else if n > 0 {
			return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)}
		}
	}
	return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"}
}

// pushIM 只投递给 IM 桥。
func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
	if p.im == nil {
		return PushResult{Pushed: false, Route: RouteNone, Detail: "no IM target"}
	}
	if err := p.im.SendProactive(ctx, msg); err != nil {
		p.log.Warn("backthink IM push failed", "error", err)
		return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()}
	}
	return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"}
}

// normalizeChannelHint 把渠道提示转小写并去空白，把同义词映射到
// 规范路由词（auto/im/web）；其它一切视为具体平台
// id（kook/telegram/...）。未知/空映射为 auto。
func normalizeChannelHint(hint string) string {
	switch strings.ToLower(strings.TrimSpace(hint)) {
	case "", "auto":
		return channelAuto
	case "im", "bridge":
		return channelIM
	case "web", "frontend", "active":
		return channelWeb
	default:
		return strings.ToLower(strings.TrimSpace(hint))
	}
}

// isSpecificChannel 报告 hint 指的是具体平台，而不是
// 路由关键词（auto/im/web）。
func isSpecificChannel(hint string) bool {
	return hint != channelAuto && hint != channelIM && hint != channelWeb
}

// annotateHint 把请求的渠道提示附加到推送结果 detail 上，
// 好让记录/面板显示她选了哪个渠道，即便路由发生
// 了降级。空白/auto 提示不加任何内容。
func annotateHint(res PushResult, hint string) PushResult {
	h := strings.ToLower(strings.TrimSpace(hint))
	if h == "" || h == channelAuto {
		return res
	}
	tag := "channel_hint=" + h
	if res.Detail == "" {
		res.Detail = tag
	} else {
		res.Detail = res.Detail + ", " + tag
	}
	return res
}

// annotateBypass 给事件驱动的推送结果打标记，使记录/面板显示它
// 跳过了 PushGate，与因限流被拒的心跳推送区分开。
func annotateBypass(res PushResult) PushResult {
	if res.Detail == "" {
		res.Detail = "event_bypass"
	} else {
		res.Detail = res.Detail + ", event_bypass"
	}
	return res
}

// rateGate 是生产用的 PushGate：一个基于
// session.RateLimiter 的滑动窗口计数器（design/11 §6 反骚扰规则），
// 结合推送后冷却与按风格（问候/提醒/分享/关心）窗口，
// 使主人绝不会被同一风格刷屏（design/12 §6 策略二）。
type rateGate struct {
	log        *slog.Logger
	limiter    *session.RateLimiter
	cooldown   time.Duration
	kindWindow time.Duration
	now        func() time.Time

	mu           sync.Mutex
	lastPushTime time.Time
	lastKind     map[PushKind]time.Time
}

// newRateGate 构建一个门：在 window 内最多放行 limit 次推送，
// 成功之后推送之间至少间隔 cooldown，且每种 PushKind 在
// kindWindow 内只允许一次。非正的 limit 或 window 得到始终放行的
// 限流器（无限频规则）；非正的 cooldown 或 kindWindow 会禁用对应的那道守护。
func newRateGate(limit int, window, cooldown, kindWindow time.Duration, log *slog.Logger) *rateGate {
	return &rateGate{
		log:        log,
		limiter:    session.NewRateLimiter(limit, window),
		cooldown:   cooldown,
		kindWindow: kindWindow,
		now:        time.Now,
		lastKind:   make(map[PushKind]time.Time),
	}
}

// Allow 实现 PushGate。已取消的 context 一律拒绝（fail closed）。
// 当最近一次成功推送仍在冷却期内，或同一风格
// 在 kindWindow 内已投递过，它就在咨询滑动窗口限流器之前拒绝。
func (g *rateGate) Allow(ctx context.Context, kind PushKind) bool {
	if err := ctx.Err(); err != nil {
		return false
	}
	now := g.now()
	g.mu.Lock()
	defer g.mu.Unlock()
	if !g.lastPushTime.IsZero() && now.Sub(g.lastPushTime) < g.cooldown {
		return false
	}
	if lt, ok := g.lastKind[kind]; ok && now.Sub(lt) < g.kindWindow {
		return false
	}
	return g.limiter.Allow()
}

// RecordPush 实现 PushGate：它记下最近成功推送时间与
// 按风格的最近时间，使冷却与风格窗口从真实投递起算。
func (g *rateGate) RecordPush(kind PushKind) {
	now := g.now()
	g.mu.Lock()
	defer g.mu.Unlock()
	g.lastPushTime = now
	g.lastKind[kind] = now
}

// buildProactiveMessage 拼装承载思考结论文本的、
// 归一化的出站消息。当结论指明具体平台时，它用请求的渠道
// 给平台打标签，否则用 backthink 的默认
// 平台。具体的活跃客户端与 IM 适配器各自对主人施加
// 自己的路由/寻址（红线：只针对主人）。
//
// 正文会再过一遍 stripInternalMarkers 作为最后一道守护，因此
// 侥幸存活过解析的结构化决策/渠道标记绝不
// 能到达面向用户的消息（只有我们的日志/记录可以携带它）。
func (b *Backthink) buildProactiveMessage(message, channelHint string) adapter.UnifiedMessage {
	platform := defaultProactivePlatform
	if h := normalizeChannelHint(channelHint); isSpecificChannel(h) {
		platform = h
	}
	return adapter.UnifiedMessage{
		Platform:    platform,
		ChannelType: adapter.ChannelDirect,
		Content:     stripInternalMarkers(message),
		ContentType: adapter.ContentText,
		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
		Timestamp:   b.now(),
	}
}