Cyrene_v2/internal/services/backthink/push.go

package backthink

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"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
	// 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 detail=%q err=%q", p.Pushed, p.Route, p.Detail, p.Err)
	}
	return fmt.Sprintf("pushed=%v route=%s detail=%q", p.Pushed, p.Route, 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. 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 may be sent now.
	Allow(ctx context.Context) bool
}

// 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 keep the default
// active-client-first, IM-fallback route; "web" targets the active client; "im"
// targets the IM bridge; a concrete platform id (kook/telegram/...) 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
}

// Pusher delivers a proactive message under the active-client-first, IM-fallback
// policy (design/12 §3), bounded by the rate gate, honoring the channel hint the
// autonomous conclusion selected.
type Pusher interface {
	// Push delivers req preferring the active client and falling back to the IM
	// bridge, 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 tries the active
// client and falls back to IM, honoring the requested channel hint.
type pusher struct {
	active ActiveClient
	im     IMAdapter
	gate   PushGate
	log    *slog.Logger
}

// 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.
func newPusher(active ActiveClient, im IMAdapter, gate PushGate, log *slog.Logger) *pusher {
	return &pusher{active: active, im: im, gate: gate, log: log}
}

// Push implements Pusher: it applies the rate gate, then resolves the channel
// hint to a route. A specific platform targets the IM bridge (tagged on the
// message) and falls back to the default route when no IM sink is wired, so the
// owner still receives the message rather than having it 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()}
	}
	if p.gate != nil && !p.gate.Allow(ctx) {
		p.log.Warn("backthink push rate limited")
		return PushResult{Pushed: false, Route: RouteRateLimited, Detail: "proactive push rate limited"}
	}

	hint := normalizeChannelHint(req.ChannelHint)
	msg := req.Message
	// A specific platform names the requested channel on the message so a
	// platform-aware IM sink can honor it; auto/web/im keep the default platform.
	if isSpecificChannel(hint) {
		msg.Platform = hint
	}

	var res PushResult
	switch hint {
	case channelWeb:
		res = p.pushActive(ctx, msg)
	case channelIM:
		res = p.pushIM(ctx, msg)
	case channelAuto:
		res = p.pushDefault(ctx, msg)
	default:
		// A specific platform targets the IM bridge for that channel. With no IM
		// sink wired (the backthink IM seam is single-sink), fall back to the
		// default route so the owner is still reached; the hint is recorded.
		res = p.pushIM(ctx, msg)
		if !res.Pushed {
			res = p.pushDefault(ctx, msg)
		}
	}
	return annotateHint(res, req.ChannelHint)
}

// pushDefault tries the active client first and falls back to the IM bridge.
func (p *pusher) pushDefault(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)}
		}
	}
	if p.im == nil {
		return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client or IM target"}
	}
	return p.pushIM(ctx, msg)
}

// 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
}

// rateGate is the production PushGate: a sliding-window counter over
// session.RateLimiter (design/11 §6 anti-spam rule).
type rateGate struct {
	log     *slog.Logger
	limiter *session.RateLimiter
}

// newRateGate builds a gate that allows at most limit pushes within window. A
// non-positive limit or window yields an always-allow gate (no rule configured).
func newRateGate(limit int, window time.Duration, log *slog.Logger) *rateGate {
	return &rateGate{log: log, limiter: session.NewRateLimiter(limit, window)}
}

// Allow implements PushGate. A canceled context denies (fail closed).
func (g *rateGate) Allow(ctx context.Context) bool {
	if err := ctx.Err(); err != nil {
		return false
	}
	return g.limiter.Allow()
}

// 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).
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:     message,
		ContentType: adapter.ContentText,
		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
		Timestamp:   b.now(),
	}
}