AskaEth
074ffc4ebf
主人 2026-10-03 拍板「注释一律中文」；gateway 至此整包注释中文化完成（本会话共 3 批： 已推库 16 个文件 + 本批剩余文件）。 校验：机械校验（非注释改动行）= 空；剥注释后代码文本逐行比对全等（不一致文件数 0）； 全树 go test ./... 58 包 ok / 0 FAIL；交叉编译绿。
312 lines
11 KiB
Go
312 lines
11 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"net/http"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
|
||
)
|
||
|
||
// 通知 feed 的限制（design/10 §3 #16）。编译期常量，因此运行路径中
|
||
// 不出现裸魔法数字（项目规范 §10）。
|
||
const (
|
||
defaultNotificationLimit = 50
|
||
maxNotificationLimit = 200
|
||
// notificationFetchFactor 加宽每来源的读取。合并页必须是并集的
|
||
// 最新 N 条，而每来源的最新 N 条总是包含它，
|
||
// 但在带 `before` 游标时某一来源可能占据最新窗口并
|
||
// 饿死另一来源 —— 该系数就是为此付出的代价。
|
||
notificationFetchFactor = 4
|
||
// notificationBodyRunes 限定 feed 中后台思考摘要的长度；全文
|
||
// 仍可在 GET /api/v1/admin/backthink 获取。
|
||
notificationBodyRunes = 200
|
||
)
|
||
|
||
// 统一 feed 的族，既用作条目的 `type`，也用作 `?type=`
|
||
// 过滤器。它们派生自组合根已有的来源
|
||
// （design/10 §3 #16：先聚合现有源为统一 feed，避免新建重模型）。
|
||
const (
|
||
// NotificationTypeSystem 是系统/安全操作：备份与恢复、
|
||
// JWT 轮换、API key 吊销、插件安装/卸载、工具策略。
|
||
NotificationTypeSystem = "system"
|
||
// NotificationTypeChannel 是渠道生命周期或连接事件
|
||
// （channel.open / close / save / reject / login）。
|
||
NotificationTypeChannel = "channel"
|
||
// NotificationTypeBackthink 是后台思考器决定发送的一次
|
||
// 主动推送。
|
||
NotificationTypeBackthink = "backthink"
|
||
// NotificationTypeAudit 是剩余的管理审计链：人格、记忆、
|
||
// 用户、RAG、会话、上传、音频。
|
||
NotificationTypeAudit = "audit"
|
||
)
|
||
|
||
// 条目 `source` 字段上携带的来源标签。
|
||
const (
|
||
notificationSourceAudit = "audit"
|
||
notificationSourceBackthink = "backthink"
|
||
// notificationIDPrefixAudit / ...Backthink 为来源自身的记录 id 加命名空间，
|
||
// 使合并 feed 的 id 稳定、无冲突、客户端可去重。
|
||
notificationIDPrefixAudit = "audit:"
|
||
notificationIDPrefixBackthink = "backthink:"
|
||
)
|
||
|
||
// AdminNotification 是统一通知 feed 的一个条目（design/10 §3
|
||
// #16）。它是既有记录 —— 审计行与后台思考记录 —— 的只读
|
||
// 投影，绝不是新的持久化模型：`id` 是 "<source>:<原生 id>"，
|
||
// `title`/`body` 派生自来源记录，`createdAt` 是 epoch
|
||
// 毫秒。
|
||
type AdminNotification struct {
|
||
ID string `json:"id"`
|
||
Type string `json:"type"`
|
||
Title string `json:"title"`
|
||
Body string `json:"body,omitempty"`
|
||
Source string `json:"source"`
|
||
CreatedAt int64 `json:"createdAt"`
|
||
}
|
||
|
||
// adminListNotifications 返回统一通知 feed（owner/admin
|
||
// 读）：管理审计链与后台思考器的主动推送，
|
||
// 最新优先合并，可按族与 `before` 游标过滤。
|
||
//
|
||
// 未注入的来源不贡献任何条目 —— feed 报告确实存在的
|
||
// 内容，而不是编造一个族 —— 因此没有后台思考服务的
|
||
// 部署就是没有 `backthink` 条目。
|
||
func (s *Server) adminListNotifications(w http.ResponseWriter, r *http.Request) {
|
||
limit := parseNotificationLimit(r)
|
||
typ := strings.TrimSpace(r.URL.Query().Get("type"))
|
||
if typ != "" && !validNotificationType(typ) {
|
||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "invalid_type",
|
||
"type must be one of system|channel|backthink|audit"))
|
||
return
|
||
}
|
||
before, hasBefore := parseNotificationBefore(r)
|
||
|
||
items, err := s.notificationItems(r.Context(), notificationFetchLimit(limit))
|
||
if err != nil {
|
||
s.log.Error("list notifications", "err", err)
|
||
writeError(w, apperr.Wrap(apperr.ErrInternal, "notifications_failed",
|
||
"failed to list notifications"))
|
||
return
|
||
}
|
||
|
||
out := make([]AdminNotification, 0, limit)
|
||
for _, it := range items {
|
||
if typ != "" && it.Type != typ {
|
||
continue
|
||
}
|
||
if hasBefore && it.CreatedAt > before {
|
||
continue
|
||
}
|
||
out = append(out, it)
|
||
if len(out) == limit {
|
||
break
|
||
}
|
||
}
|
||
writeJSON(w, http.StatusOK, out)
|
||
}
|
||
|
||
// notificationItems 读取两个来源并返回它们的合并结果，最新优先。
|
||
// 来源失败会被返回（绝不被吞进静默变短的 feed），因此
|
||
// 端点返回明确的错误，而不是假装没有事件。
|
||
func (s *Server) notificationItems(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||
auditItems, err := s.notificationsFromAudit(ctx, fetch)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
btItems, err := s.notificationsFromBackthink(ctx, fetch)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
items := append(auditItems, btItems...)
|
||
sort.SliceStable(items, func(i, j int) bool {
|
||
if items[i].CreatedAt != items[j].CreatedAt {
|
||
return items[i].CreatedAt > items[j].CreatedAt
|
||
}
|
||
// id 降序为共享同一毫秒的条目保持确定性顺序
|
||
// （游标可能把一毫秒拆到两页）。
|
||
return items[i].ID > items[j].ID
|
||
})
|
||
return items, nil
|
||
}
|
||
|
||
// notificationsFromAudit 把管理审计链（audit_logs，由
|
||
// NewSQLiteAuditStore 持久化）投影为 feed 条目。回退的内存存储
|
||
// 只保留本进程真正追加过的记录，因此未接线的存储得到
|
||
// 空族，而不是编造条目。
|
||
func (s *Server) notificationsFromAudit(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||
if s.obs == nil || s.obs.audit == nil {
|
||
return nil, nil
|
||
}
|
||
records, err := s.obs.audit.List(ctx, fetch)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("gateway: notification audit source: %w", err)
|
||
}
|
||
out := make([]AdminNotification, 0, len(records))
|
||
for _, rec := range records {
|
||
out = append(out, AdminNotification{
|
||
ID: notificationIDPrefixAudit + strconv.FormatInt(rec.ID, 10),
|
||
Type: notificationTypeOfAudit(rec.What),
|
||
Title: rec.What,
|
||
Body: notificationAuditBody(rec),
|
||
Source: notificationSourceAudit,
|
||
CreatedAt: notificationAuditAt(rec.When),
|
||
})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// notificationsFromBackthink 把后台思考器的记录投影为 feed
|
||
// 条目。只有决定推送的记录才是通知（普通思考
|
||
// 记录不是需要告知主人的事件）；每条都携带
|
||
// 投递状态，因此 feed 绝不声称未投递的推送已发出。
|
||
func (s *Server) notificationsFromBackthink(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||
if s.backthink == nil {
|
||
return nil, nil
|
||
}
|
||
records, err := s.backthink.List(ctx, fetch)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("gateway: notification backthink source: %w", err)
|
||
}
|
||
out := make([]AdminNotification, 0, len(records))
|
||
for _, rec := range records {
|
||
if !rec.ShouldPush {
|
||
continue
|
||
}
|
||
out = append(out, AdminNotification{
|
||
ID: notificationIDPrefixBackthink + strconv.FormatInt(rec.ID, 10),
|
||
Type: NotificationTypeBackthink,
|
||
Title: backthinkNotificationTitle(rec),
|
||
Body: clipRunes(strings.TrimSpace(rec.Conclusion), notificationBodyRunes),
|
||
Source: notificationSourceBackthink,
|
||
CreatedAt: rec.At.UnixMilli(),
|
||
})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// notificationTypeOfAudit 把审计操作归类到 feed 族：
|
||
// 渠道生命周期是 `channel`，系统/插件/策略操作是
|
||
// `system`；其余保持 `audit`。
|
||
func notificationTypeOfAudit(what string) string {
|
||
switch {
|
||
case strings.HasPrefix(what, "channel."):
|
||
return NotificationTypeChannel
|
||
case strings.HasPrefix(what, "system."),
|
||
strings.HasPrefix(what, "plugin."),
|
||
strings.HasPrefix(what, "policy."):
|
||
return NotificationTypeSystem
|
||
default:
|
||
return NotificationTypeAudit
|
||
}
|
||
}
|
||
|
||
// notificationAuditBody 从审计条目的 who/what/result
|
||
// 链推导其 body，例如 "owner:owner · ok"。
|
||
func notificationAuditBody(rec AuditRecord) string {
|
||
parts := make([]string, 0, 2)
|
||
if who := strings.TrimSpace(rec.Who); who != "" {
|
||
parts = append(parts, who)
|
||
}
|
||
if result := strings.TrimSpace(rec.Result); result != "" {
|
||
parts = append(parts, result)
|
||
}
|
||
return strings.Join(parts, " · ")
|
||
}
|
||
|
||
// notificationAuditAt 把审计记录的 RFC3339 时间戳解析为 epoch
|
||
// 毫秒。无法解析的时间戳得到 0：条目仍然出现（最旧
|
||
// 优先），而不是被丢弃或被盖上一个编造的时间。
|
||
func notificationAuditAt(when string) int64 {
|
||
ts, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(when))
|
||
if err != nil {
|
||
return 0
|
||
}
|
||
return ts.UnixMilli()
|
||
}
|
||
|
||
// backthinkNotificationTitle 用触发原因与投递状态标注一条主动
|
||
// 推送，使 feed 区分「已发送」与「已决定但未
|
||
// 投递」（route none / 无在线客户端 / 推送失败）。
|
||
func backthinkNotificationTitle(rec backthink.Record) string {
|
||
state := "未送达"
|
||
if rec.PushResult.Pushed {
|
||
state = "已送达"
|
||
}
|
||
if trigger := strings.TrimSpace(rec.Trigger); trigger != "" {
|
||
return "主动推送（" + trigger + " · " + state + "）"
|
||
}
|
||
return "主动推送（" + state + "）"
|
||
}
|
||
|
||
// clipRunes 把 s 截断到至多 max 个 rune，被截断时追加省略号。
|
||
// 它绝不切开多字节 rune。
|
||
func clipRunes(s string, max int) string {
|
||
if max <= 0 || utf8.RuneCountInString(s) <= max {
|
||
return s
|
||
}
|
||
runes := []rune(s)
|
||
return string(runes[:max]) + "…"
|
||
}
|
||
|
||
// parseNotificationLimit 解析 ?limit= 查询参数，夹取到
|
||
// [1, maxNotificationLimit]，缺失或非法时选用
|
||
// defaultNotificationLimit 作为默认值。
|
||
func parseNotificationLimit(r *http.Request) int {
|
||
n, err := strconv.Atoi(r.URL.Query().Get("limit"))
|
||
if err != nil || n <= 0 {
|
||
return defaultNotificationLimit
|
||
}
|
||
if n > maxNotificationLimit {
|
||
return maxNotificationLimit
|
||
}
|
||
return n
|
||
}
|
||
|
||
// parseNotificationBefore 解析 ?before= epoch 毫秒游标。该
|
||
// 游标是包含式的（返回 createdAt <= before 的条目）：来源
|
||
// 解析到毫秒，因此页边界可能落在一毫秒之内，
|
||
// 而客户端按 id 去重。游标缺失或畸形意味着「从
|
||
// 最新条目开始」。
|
||
func parseNotificationBefore(r *http.Request) (int64, bool) {
|
||
raw := strings.TrimSpace(r.URL.Query().Get("before"))
|
||
if raw == "" {
|
||
return 0, false
|
||
}
|
||
ms, err := strconv.ParseInt(raw, 10, 64)
|
||
if err != nil || ms <= 0 {
|
||
return 0, false
|
||
}
|
||
return ms, true
|
||
}
|
||
|
||
// notificationFetchLimit 把页请求加宽为每来源读取窗口，
|
||
// 上限为 maxObsLimit（审计存储自身的上限），且绝不低于该页。
|
||
func notificationFetchLimit(limit int) int {
|
||
fetch := limit * notificationFetchFactor
|
||
if fetch > maxObsLimit {
|
||
fetch = maxObsLimit
|
||
}
|
||
if fetch < limit {
|
||
fetch = limit
|
||
}
|
||
return fetch
|
||
}
|
||
|
||
// validNotificationType 报告 t 是否是一个 feed 族名。未知族
|
||
// 以 400 拒绝，而不是以静默为空的 feed 作答。
|
||
func validNotificationType(t string) bool {
|
||
switch t {
|
||
case NotificationTypeSystem, NotificationTypeChannel, NotificationTypeBackthink, NotificationTypeAudit:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|