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 identifies how a proactive message was actually delivered, for the
// monitor panel and the push result record.
type Route string

const (
	// RouteNone means no push was attempted or no target was available.
	RouteNone Route = "none"
	// RouteRateLimited means the push was refused by the rate limiter.
	RouteRateLimited Route = "rate_limited"
	// RouteActiveClient means the message went to an active web/native client.
	RouteActiveClient Route = "active_client"
	// RouteIMBridge means the message fell back to the IM bridge.
	RouteIMBridge Route = "im_bridge"
)

// Channel-hint tokens recognized by the push router. They come from the
// conclusion's [渠道:...] marker (parseProactive) and select where a proactive
// message is delivered. An empty/unknown hint is treated as auto. A concrete
// platform id (kook/telegram/...) is a value other than these three.
const (
	channelAuto = "auto"
	channelIM   = "im"
	channelWeb  = "web"
	// defaultProactivePlatform is the Platform the outbound proactive message
	// carries when the conclusion names no specific channel.
	defaultProactivePlatform = "cyrene-backthink"
)

// String returns the route as a plain label.
func (r Route) String() string { return string(r) }

// PushResult is the outcome of one proactive push attempt, recorded with each
// thinking pass (design/12 §4).
type PushResult struct {
	// Pushed reports whether the message was delivered.
	Pushed bool
	// Route is the delivery route taken.
	Route Route
	// Bypassed reports whether this push was event-driven and skipped the PushGate
	// (a real event, "该发就发"). A heartbeat push is gate-governed and never
	// bypassed.
	Bypassed bool
	// Detail is a short human-readable note (e.g. client count, decision reason).
	Detail string
	// Err is the underlying error text when delivery failed, or that of the
	// provider that decided not to push.
	Err string
}

// String returns a concise summary for logs.
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 delivers a proactive message to every currently active web/native
// client and reports how many received it. It is the preferred push target. The
// gateway WebSocket hub implements this in production; tests inject a mock.
type ActiveClient interface {
	// SendProactive pushes msg to each active client and returns the number of
	// clients that accepted it.
	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error)
}

// IMAdapter is the IM bridge fallback: it pushes a proactive message over an IM
// channel. Every adapter.PlatformAdapter satisfies this subset, so the service
// depends on the narrow interface rather than the concrete adapter type.
type IMAdapter interface {
	// SendProactive pushes msg to the owner's IM channel.
	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error
}

// PushGate decides whether a proactive push may proceed right now, applying the
// rate limits of design/11 §6 plus a per-flavor limit and a post-push cooldown. It
// is a seam over channel/session's ProactivePolicy/RateLimiter so the service can
// be tested with a mock.
type PushGate interface {
	// Allow reports whether a proactive push of the given kind may be sent now.
	Allow(ctx context.Context, kind PushKind) bool
	// RecordPush notes a successful delivery so the cooldown/per-kind windows
	// start from a real successful push, not a merely-attempted one.
	RecordPush(kind PushKind)
}

// PushRequest carries one proactive push attempt: the outbound message plus the
// channel target the autonomous conclusion asked for (design/20 §2 P2).
// ChannelHint is the lowercased [渠道:...] token: ""/auto pick the default (the
// owner mandate: ONE channel, chosen by her / her most recent platform / the
// whitelist default); "web" targets the active client; "im" targets the IM bridge;
// a concrete platform id (kook/telegram/weixin-ilink/...) targets the IM bridge
// with the message tagged for that platform.
type PushRequest struct {
	// Message is the normalized outbound proactive message.
	Message adapter.UnifiedMessage
	// ChannelHint is the requested target channel, "" when she did not choose one.
	ChannelHint string
	// TriggerKind records what woke this pass. An event-driven push bypasses the
	// PushGate (owner mandate: "该发就发"); a heartbeat stays gate-governed.
	TriggerKind TriggerKind
}

// Pusher delivers a proactive message to ONE target channel. It is bounded by the
// rate gate and honors the channel hint the autonomous conclusion selected: "web"
// goes to the active client, everything else goes over the IM bridge a single time,
// where the bridge picks the one channel (her choice → her most recent platform →
// whitelist default, WeChat included).
type Pusher interface {
	// Push delivers req to a single target channel, per the channel hint. It never
	// blocks on a slow target.
	Push(ctx context.Context, req PushRequest) PushResult
}

// pusher is the concrete Pusher: it consults the gate, then routes the message to
// exactly one target — the active client for a "web" hint, otherwise the IM bridge.
// After the local delivery it mirrors the message to the optional push outlet
// (self-hosted service → Getui fallback) without blocking that delivery.
type pusher struct {
	active   ActiveClient
	im       IMAdapter
	gate     PushGate
	exporter PushExporter
	log      *slog.Logger
	// exportDone, when non-nil, receives one value after each asynchronous export
	// completes. It exists so tests can deterministically observe the hook; the
	// send is non-blocking, so a full channel never stalls the exporter.
	exportDone chan struct{}
}

// newPusher builds a Pusher from the optional active-client sink, IM sink and
// rate gate. Any of them may be nil; a nil target simply makes that route
// unavailable. The push outlet is attached separately with withExporter.
func newPusher(active ActiveClient, im IMAdapter, gate PushGate, log *slog.Logger) *pusher {
	return &pusher{active: active, im: im, gate: gate, log: log}
}

// withExporter attaches the push outlet hook and returns the pusher for fluent
// use. A nil exporter leaves the hook disabled.
func (p *pusher) withExporter(exporter PushExporter) *pusher {
	p.exporter = exporter
	return p
}

// Push implements Pusher: it applies the rate gate, then routes the message to a
// single target. A "web" hint reaches the active web/native client; every other
// hint ("im", "auto", or a concrete platform) goes over the IM bridge once, where
// the bridge selects one owner channel (her [渠道:...] choice, then the platform she
// used most recently, then the whitelist default). The IM bridge is primary, so a
// push never fans out to several channels. When no IM sink is wired the active
// client is the honest fallback, so the owner is still reached rather than the
// message being dropped.
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)
	// A real event-driven push bypasses the PushGate: it should go out as-is ("该发就发"),
	// independent of the heartbeat throttle (60min≤1 / cooldown / per-flavor window).
	// Only the periodic heartbeat stays gate-governed, so the owner is never spammed
	// by an empty reflection. Same-event dedup is applied upstream in 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"/a concrete platform all travel over the IM bridge once. A
		// generic hint hands the sentinel platform so the bridge falls through to its
		// recent-active / default selection; a concrete hint is honored when the
		// channel is whitelisted and connected.
		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)
		}
	}
	// Only a real delivery advances the cooldown/per-kind windows (and only for a
	// gate-governed push), so an attempted-but-failed push does not suppress a later
	// genuine one, and an event push never contributes to the heartbeat throttle.
	if res.Pushed && p.gate != nil && !eventDriven {
		p.gate.RecordPush(kind)
	}
	// D1: mirror the proactive message to the external push outlet (self-hosted
	// unified push service → Getui fallback). Fire-and-forget: a slow or failing
	// outlet never delays or breaks the local delivery above, and a rate-limited
	// push returned before reaching here.
	p.exportAsync(msg)
	res = annotateHint(res, req.ChannelHint)
	if eventDriven {
		res.Bypassed = true
		res = annotateBypass(res)
	}
	return res
}

// exportAsync reports msg to the push outlet on a detached, bounded goroutine so
// the proactive path never blocks on the outlet. A nil exporter (hook disabled)
// or a nil outlet result is a no-op; a delivery miss or error is only logged.
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 delivers only to the active web/native client.
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 delivers only to the IM bridge.
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 lowercases and trims a channel hint, mapping synonyms to
// the canonical route tokens (auto/im/web); anything else is a concrete platform
// id (kook/telegram/...). Unknown/blank maps to 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 reports whether hint names a concrete platform rather than a
// routing keyword (auto/im/web).
func isSpecificChannel(hint string) bool {
	return hint != channelAuto && hint != channelIM && hint != channelWeb
}

// annotateHint appends the requested channel hint to the push result detail so
// the record/panel surfaces which channel she chose, even when the route fell
// back. Blank/auto hints add nothing.
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 tags an event-driven push result so the record/panel shows it
// skipped the PushGate, distinct from a heartbeat push that was rate-limited.
func annotateBypass(res PushResult) PushResult {
	if res.Detail == "" {
		res.Detail = "event_bypass"
	} else {
		res.Detail = res.Detail + ", event_bypass"
	}
	return res
}

// rateGate is the production PushGate: a sliding-window counter over
// session.RateLimiter (design/11 §6 anti-spam rule) combined with a post-push
// cooldown and a per-flavor (问候/提醒/分享/关心) window so the owner is never
// spammed with the same style (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 builds a gate that allows at most limit pushes within window, keeps
// pushes cooldown apart after a success, and limits each PushKind to one push per
// kindWindow. A non-positive limit or window yields an always-allow limiter (no
// rate rule); a non-positive cooldown or kindWindow disables that specific guard.
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 implements PushGate. A canceled context denies (fail closed). It refuses
// when the last successful push is inside the cooldown, or when the same flavor
// was delivered within kindWindow, before consulting the sliding-window limiter.
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 implements PushGate: it stamps the last successful push time and the
// per-flavor last time so the cooldown and kind window start from a real delivery.
func (g *rateGate) RecordPush(kind PushKind) {
	now := g.now()
	g.mu.Lock()
	defer g.mu.Unlock()
	g.lastPushTime = now
	g.lastKind[kind] = now
}

// buildProactiveMessage assembles the normalized outbound message carrying a
// think conclusion's clean body. It tags the platform with the requested channel
// when the conclusion names a concrete platform, otherwise the default backthink
// platform. The concrete active-client and IM adapters apply their own
// routing/targeting to the owner (red line: owner only).
//
// The body is run through stripInternalMarkers as a final guard so a structured
// decision/channel marker that survived parsing can never reach the user-facing
// message (only our logs/records may carry it).
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(),
	}
}