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