Cyrene_v2/internal/services/backthink/push_outlet.go

package backthink

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"strings"
	"time"

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

// 推送出口（D1，design/12 §4/§6）：主动消息被镜像到
// 外部推送服务，使主人的移动端即便在没有
// WebSocket 客户端或 IM 渠道在线时也能被触达。按设计投递分两级：
//
//  1. 主人的自托管统一推送服务（主级；PUSH_EXPORTER_URL），
//  2. 个推（Getui）作为自托管服务不可用时的降级。
//
// 第 2 级尚未接入 SDK，因此主级不可用时只记一条明确的
// "Getui fallback not integrated" 备注，并上报未投递，而不
// 假装消息已发出。该钩子是发后不管：它绝不延迟
// 或破坏既有的活跃客户端 / IM 投递。

// defaultExportTimeout 约束单次导出尝试，因此挂死的端点无法永久
// 占住一个 goroutine。
const defaultExportTimeout = 5 * time.Second

// maxExportResponseBytes 限制连接复用前从出口响应体里
// 抽掉多少数据，因此恶意或有缺陷的端点无法把一个
// 无界的响应体流进内存。
const maxExportResponseBytes = 4 << 10

// PushExporter 把一条主动消息上报给外部推送出口。Export
// 必须并发安全且必须尊重 ctx。nil 错误配上
// delivered=false 表示出口无法投递（例如降级
// 级尚未接线）；调用方把它当作一次记账的未命中，绝不当作本地
// 推送路径的失败。
type PushExporter interface {
	Export(ctx context.Context, msg adapter.UnifiedMessage) (delivered bool, err error)
}

// PushOutletConfig 携带推送出口的接线。Enabled=false 完全关闭
// 该钩子（NewPushOutlet 返回 nil）。URL/Token/Timeout 描述
// 自托管主服务；Getui* 字段保留给延后的
// 降级级，目前尚未消费。
type PushOutletConfig struct {
	// Enabled 打开出口钩子。
	Enabled bool
	// URL 是自托管统一推送服务端点（HTTP POST JSON）。
	// 为空则主级保持未配置。
	URL string
	// Token 以 "Authorization: Bearer <token>" 发送。它绝不写日志。
	Token string
	// Timeout 约束单次导出尝试。为零用 defaultExportTimeout。
	Timeout time.Duration
	// Log 是基础日志器。nil 则用丢弃式日志器。
	Log *slog.Logger
}

// NewPushOutlet 由 cfg 构建两级推送出口。钩子关闭时它返回 nil，
// 因此推送器直接跳过导出步骤。启用时，
// 主级是自托管 HTTP 端点（仅当
// URL 已设置时存在），降级级是个推，尚未接入。
func NewPushOutlet(cfg PushOutletConfig) PushExporter {
	if !cfg.Enabled {
		return nil
	}
	log := logger.WithSource(baseLog(cfg.Log), "services/backthink/push_outlet")
	timeout := cfg.Timeout
	if timeout <= 0 {
		timeout = defaultExportTimeout
	}
	var primary PushExporter
	if url := strings.TrimSpace(cfg.URL); url != "" {
		primary = &httpPushExporter{
			url:    url,
			token:  cfg.Token,
			client: &http.Client{Timeout: timeout},
		}
	} else {
		log.Warn("push outlet enabled but PUSH_EXPORTER_URL is empty; only the (unwired) Getui fallback remains")
	}
	log.Info("push outlet configured", "primary", primary != nil, "fallback", "getui (not integrated)")
	return &pushOutlet{primary: primary, log: log}
}

// pushOutlet 是两级导出器：先自托管服务，再个推。
// 在个推 SDK 接入前降级级为 nil，因此主级未命中时
// 只记日志并上报未投递，而不是静默丢弃。
type pushOutlet struct {
	primary  PushExporter
	fallback PushExporter
	log      *slog.Logger
}

// Export 实现 PushExporter。它先试主级，主级
// 未投递时再试降级级。未接降级级时它记录待
// 接入状态并返回 (false, nil)：调用方把这当作未命中，
// 绝不当作让主动推送失败的理由。
func (o *pushOutlet) Export(ctx context.Context, msg adapter.UnifiedMessage) (bool, error) {
	if o.primary != nil {
		delivered, err := o.primary.Export(ctx, msg)
		if err == nil && delivered {
			return true, nil
		}
		if err != nil {
			o.log.Warn("push outlet primary failed", "error", err)
		} else {
			o.log.Warn("push outlet primary did not deliver")
		}
	}
	if o.fallback != nil {
		return o.fallback.Export(ctx, msg)
	}
	o.log.Info("push outlet not delivered; getui fallback not integrated",
		"hint", "wire the Getui SDK + GETUI_* credentials to enable the fallback tier")
	return false, nil
}

// pushExportPayload 是 POST 到自托管推送服务的 JSON 体。它
// 刻意小而稳定：设备寻址归该服务，消息
// 只携带渲染与去重所需的内容。
type pushExportPayload struct {
	// Source 标识生产者，好让服务能路由/过滤。
	Source string `json:"source"`
	// Title 是简短显示标题（角色名）。
	Title string `json:"title"`
	// Content 是主动消息正文（内部决策标记
	// 已由 buildProactiveMessage 剥掉）。
	Content string `json:"content"`
	// Platform 是出站平台标签（cyrene-backthink 或具体的 IM
	// 平台）。
	Platform string `json:"platform"`
	// Timestamp 是消息时间，Unix 毫秒。
	Timestamp int64 `json:"timestamp"`
}

// httpPushExporter 把消息 POST 到自托管推送服务。2xx
// 响应表示已投递；其它任何状态或传输错误都算错误。
// token 只出现在 Authorization 头里，绝不进入日志。
type httpPushExporter struct {
	url    string
	token  string
	client *http.Client
}

// Export 实现 PushExporter，走 HTTP POST JSON。
func (e *httpPushExporter) Export(ctx context.Context, msg adapter.UnifiedMessage) (bool, error) {
	body, err := json.Marshal(pushExportPayload{
		Source:    "cyrene/backthink",
		Title:     msg.Sender.DisplayName,
		Content:   msg.Content,
		Platform:  msg.Platform,
		Timestamp: msg.Timestamp.UnixMilli(),
	})
	if err != nil {
		return false, fmt.Errorf("push outlet: marshal payload: %w", err)
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.url, bytes.NewReader(body))
	if err != nil {
		return false, fmt.Errorf("push outlet: build request: %w", e.redact(err))
	}
	req.Header.Set("Content-Type", "application/json")
	if e.token != "" {
		req.Header.Set("Authorization", "Bearer "+e.token)
	}
	resp, err := e.client.Do(req)
	if err != nil {
		return false, fmt.Errorf("push outlet: post: %w", e.redact(err))
	}
	defer func() { _ = resp.Body.Close() }()
	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxExportResponseBytes))
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return false, fmt.Errorf("push outlet: unexpected status %d", resp.StatusCode)
	}
	return true, nil
}

// redact 从传输错误里移除端点 URL，因此 URL 中内嵌的
// 秘密（例如查询参数里的 token）绝不进入日志行。
func (e *httpPushExporter) redact(err error) error {
	if err == nil || e.url == "" {
		return err
	}
	msg := strings.ReplaceAll(err.Error(), e.url, "[push-outlet-url]")
	return fmt.Errorf("%s", msg)
}