Cyrene_v2/internal/services/backthink/push.go

package backthink

import (
	"context"
	"fmt"
	"log/slog"
	"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"
)

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

// Pusher delivers a proactive message under the active-client-first, IM-fallback
// policy (design/12 §3), bounded by the rate gate.
type Pusher interface {
	// Push delivers msg, preferring the active client and falling back to the IM
	// bridge. It never blocks on a slow target.
	Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult
}

// pusher is the concrete Pusher: it consults the gate, then tries the active
// client and falls back to IM.
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.
func (p *pusher) Push(ctx context.Context, msg adapter.UnifiedMessage) 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"}
	}

	// Active client first.
	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)}
		}
	}

	// IM fallback.
	if p.im != nil {
		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"}
	}

	return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client or IM target"}
}

// 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. It is intentionally minimal: the concrete active-client and
// IM adapters apply their own routing/targeting to the owner.
func (b *Backthink) buildProactiveMessage(conclusion string) adapter.UnifiedMessage {
	return adapter.UnifiedMessage{
		Platform:    "cyrene-backthink",
		ChannelType: adapter.ChannelDirect,
		Content:     conclusion,
		ContentType: adapter.ContentText,
		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
		Timestamp:   b.now(),
	}
}