AskaEth
4c9cabce1b
P2-1（行为缺陷）：窗口准入此前只在 prepareInbound 里算 `solo`，而 prepare 发生在 **窗口等待之后** —— 平台帧（action / 系统通知 / 我们自己的 message_sent 回执）照样 进窗、照样把滑动计时重置：一条回执就能把这一批的静默期一直续住（群友说了三句话， 她因为「撤回事件」迟迟不开口）。这与任务书「窗口只收 chat/image/voice/file」相反。 - `laneWindowAdmits(msg)` 成为**唯一准入判据**，只看结构化事实（ContentType + isSystemNoticeMessage，绝不解析正文）； - `splitWindowItems` 在窗口等待阶段把到达的条目分成「进窗口」与「不进窗口」两半； - 非准入帧**不建窗口、不重置计时**（首条即到达即发车），窗口中途到达的只收进 solos， 这一批发车后**原样回插队首**各自单跑（保序、不合并、不丢弃）； - prepareInbound 里的 `solo` 保留为第二道保险（适配器将来把平台事件标成 text 时， 仍会在组批前被切出去）。 P2-2（硬约束）：lane.go 682 行 → 按职责纯搬移成 5 个文件（最大 246 行）： lane.go（车道类型与注册表）/ lane_queue.go（入队与取队列）/ lane_worker.go（单消费者与回合槽）/ lane_drop.go（丢弃与停机诚实）/ lane_owner.go（主人私聊判据与车道键）。语义零改动：去注释后的声明集合 （34 个顶层声明，含函数体/字段/常量值）逐字节全等；既有用例一条未改。 顺带：把两处「IsOwner() && !IsGroupChat()」的注释改成代码真正用的判据 （ownerPrivateLane = !msg.IsGroup() && actor.IsOwner()），并记下这处 P3 分叉。 验证：全树 build/vet/test 全绿（73 包 ok）；三包 -race 全绿； 新增 4 条准入用例在「准入失效」变异下三条必红、对照组保持绿。
251 lines
12 KiB
Go
251 lines
12 KiB
Go
package gateway
|
||
|
||
// 本文件承载**渠道静默窗口合并**（M23）的管理读写端点：
|
||
//
|
||
// GET /api/v1/admin/channels/coalesce 读：盘上的窗口参数 + 取值规则 + 只读事实
|
||
// PUT /api/v1/admin/channels/coalesce 写：三态补丁 + 原子落盘 + 热生效 + 审计
|
||
//
|
||
// 它回答的是主人的原话「要让群聊聊天很自然，很流畅的做法」：群里「一鼓作气打三句」
|
||
// 不该让她回三次 —— 但「等多久 / 最多攒多少」这类参数只该在面板上调，
|
||
// 不该为改一个 2.0s → 3.0s 去 SSH 改文件（与 memory-tidy 调优同一取向）。
|
||
//
|
||
// 四条口径（每条都在 admin_coalesce_test.go 里钉住）：
|
||
//
|
||
// 1. **三态**：键**缺位** ⇒ 不动；**合法值** ⇒ 采用；**非法值** ⇒ `400 invalid_payload`
|
||
// 且**盘上一个字节都不动**（校验全部发生在落盘之前，见
|
||
// runtimeconfig.ValidateCoalescePatch）。
|
||
// 2. **写端点带审计**（动作 `channels.coalesce`），`writeAuth`（owner|admin）——
|
||
// 与 `admin_memory_tuning` 同档：它写的是**面板配置文件**，不是身份/凭据面，
|
||
// 因此不是 owner-only；也不给 viewer（它是渠道调优，不是监控投影）。
|
||
// 3. **热生效**：桥每次发车前从**同一个** store 实例读生效值，因此保存后下一次发车
|
||
// 就用新值 —— 响应里的 `hot:true` 是事实，不是承诺。
|
||
// 4. **主人私聊是只读事实**：车道类型由 actor 元数据决定（判据 `ownerPrivateLane`：
|
||
// `!msg.IsGroup() && actor.IsOwner()` ⇒ direct 车道、**没有窗口对象、拿不到定时器
|
||
// 工厂**），本端点**没有任何键**能翻转它。
|
||
// `ownerPrivate:"never"` 就是这条事实的下发形状，面板据此只显示一行说明、不给开关。
|
||
//
|
||
// 零新 `.env` 键（design/13 §2.1）：配置落在 `$DATA_DIR/config/coalesce.json`。
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
|
||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
)
|
||
|
||
const (
|
||
// coalesceLogSource 是本子功能的日志来源（模块/子功能可追溯，AGENTS §10）。
|
||
coalesceLogSource = "gateway/channels-coalesce"
|
||
// coalesceAuditAction 是写端点落审计的动作名（与 channels.* 同族命名）。
|
||
coalesceAuditAction = "channels.coalesce"
|
||
// coalesceMaxBodyBytes 是写端点请求体的上限：这是一份给人改的小配置
|
||
// （6 个标量键 + 少量会话覆盖），正常请求体不到 1 KiB。
|
||
coalesceMaxBodyBytes = 32 << 10
|
||
// coalesceOwnerPrivateFact 是「主人私聊不受此设置影响」的**机器可读**形式。
|
||
// 它不是开关值：没有任何请求体字段能把它改成别的（改车道类型要改 actor 判定）。
|
||
coalesceOwnerPrivateFact = "never"
|
||
)
|
||
|
||
// CoalesceConfigStore 是「渠道静默窗口配置」的可选能力：生产装配的 `s.channels`
|
||
// 就是 cmd/cyrene 的 `*runtimeChannels`，它背后持有**与 bridge 同一个**
|
||
// `runtimeconfig.CoalesceStore`（因此写进内存即热生效）。
|
||
//
|
||
// 用可选接口而不是往 Deps/Server 加字段：渠道调优本来就在这个依赖面上，
|
||
// 加字段会让「谁持有什么」多一条没人复得到的路径（与 gateway 其它可选能力同一手法）。
|
||
type CoalesceConfigStore interface {
|
||
// CoalescePath 返回配置文件落点（面板如实回显改的是哪个文件）。
|
||
CoalescePath() string
|
||
// CoalescePresent 报告配置文件是否已存在（false = 还没写过，运行期正走代码默认）。
|
||
CoalescePresent() bool
|
||
// CoalesceSnapshot 返回盘上那份原值（未 clamp，供面板回显「你写进去的是什么」）。
|
||
CoalesceSnapshot() runtimeconfig.CoalesceFile
|
||
// ApplyCoalesce 应用三态补丁并原子落盘。返回是否真的改动了盘上的字节。
|
||
ApplyCoalesce(patch runtimeconfig.CoalescePatch) (bool, error)
|
||
}
|
||
|
||
// AdminCoalescePolicy 是一组窗口参数（生效值）。
|
||
type AdminCoalescePolicy struct {
|
||
Enabled bool `json:"enabled"`
|
||
WindowMs int `json:"windowMs"`
|
||
MaxWaitMs int `json:"maxWaitMs"`
|
||
MaxBatchMessages int `json:"maxBatchMessages"`
|
||
ApplyToStrangers bool `json:"applyToStrangers"`
|
||
StrangerWindowMs int `json:"strangerWindowMs"`
|
||
}
|
||
|
||
// AdminCoalesceRules 是取值规则（随读下发 ⇒ 面板表单与后端校验不可能两套规则）。
|
||
type AdminCoalesceRules struct {
|
||
// WindowMs 是允许区间（毫秒）；越界会被后端 clamp 并留 WARN，而不是拒绝。
|
||
WindowMs [2]int `json:"windowMs"`
|
||
// MaxWaitMs 的下界**跟随窗口**（因此 [0] 是窗口下界、[1] 是绝对上限）。
|
||
MaxWaitMs [2]int `json:"maxWaitMs"`
|
||
// MaxBatchMessages 是单批条数区间（超出不是丢弃，是拆成下一批）。
|
||
MaxBatchMessages [2]int `json:"maxBatchMessages"`
|
||
// Clamp 说明越界值的处置（"clamp"），使面板不必猜。
|
||
Clamp string `json:"clamp"`
|
||
// FlushReasons 是发车原因的闭集（面板/文档据此解释日志里的 flush_reason）。
|
||
FlushReasons []string `json:"flushReasons"`
|
||
}
|
||
|
||
// AdminCoalesceView 是 GET / PUT `/admin/channels/coalesce` 的 data。
|
||
type AdminCoalesceView struct {
|
||
// Path 是面板配置文件的落点（与 channels.json 同一棵配置树）。
|
||
Path string `json:"path"`
|
||
// Present 报告那份文件是否已存在（false = 还没写过，运行期正走代码默认）。
|
||
Present bool `json:"present"`
|
||
// Hot 报告这份配置是否**热生效**（桥每次发车前重读同一 store ⇒ true）。
|
||
Hot bool `json:"hot"`
|
||
// Default 是**按代码默认 + 盘上 default 段**归一后的生效值（面板的「当前值」）。
|
||
Default AdminCoalescePolicy `json:"default"`
|
||
// Patch 是盘上 default 段的**原值**：键缺位 ⇒ 不出现（= 跟随代码默认）。
|
||
// 有了它，面板的「留空 ⇒ 跟随默认 2000ms」才是照实回显而不是猜。
|
||
Patch map[string]any `json:"patch"`
|
||
// Conversations 是每会话覆盖的**生效值**（键 = `platform|channel`，可读形态）。
|
||
Conversations map[string]AdminCoalescePolicy `json:"conversations"`
|
||
// Rules 是取值规则；OwnerPrivate 是那条只读事实（永远 "never"）。
|
||
Rules AdminCoalesceRules `json:"rules"`
|
||
OwnerPrivate string `json:"ownerPrivate"`
|
||
}
|
||
|
||
// adminCoalesceRead 是 GET 的处理函数：只读、不审计（与 /admin/config-facts 同档）。
|
||
func (s *Server) adminCoalesceRead(w http.ResponseWriter, _ *http.Request) {
|
||
store, ok := s.coalesceStore()
|
||
if !ok {
|
||
writeError(w, apperr.Wrap(apperr.ErrInternal, "coalesce_unavailable",
|
||
"渠道静默窗口配置面未接线：面板配置文件落点无法确定"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, buildCoalesceView(store))
|
||
}
|
||
|
||
// adminCoalesceWrite 是 PUT 的处理函数：解析 → 校验 → 落盘 → 回显真值 → 审计。
|
||
//
|
||
// 顺序是刻意的（口径 1）：**任何**校验失败都在落盘之前返回 400，
|
||
// 因此「非法值 ⇒ 盘上一个字节都不动」是结构性的，不是靠回滚。
|
||
func (s *Server) adminCoalesceWrite(w http.ResponseWriter, r *http.Request) {
|
||
store, ok := s.coalesceStore()
|
||
if !ok {
|
||
s.auditClaims(r, coalesceAuditAction, "error")
|
||
writeError(w, apperr.Wrap(apperr.ErrInternal, "coalesce_unavailable",
|
||
"渠道静默窗口配置面未接线：面板配置文件落点无法确定"))
|
||
return
|
||
}
|
||
body, err := io.ReadAll(io.LimitReader(r.Body, coalesceMaxBodyBytes))
|
||
if err != nil {
|
||
s.auditClaims(r, coalesceAuditAction, "error")
|
||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "读取请求体失败"))
|
||
return
|
||
}
|
||
patch, err := parseCoalescePatch(body)
|
||
if err != nil {
|
||
s.auditClaims(r, coalesceAuditAction, "error")
|
||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "invalid_payload", "%v", err))
|
||
return
|
||
}
|
||
wrote := false
|
||
if !patch.Empty() {
|
||
wrote, err = store.ApplyCoalesce(patch)
|
||
if err != nil {
|
||
s.auditClaims(r, coalesceAuditAction, "error")
|
||
logger.WithSource(s.log, coalesceLogSource).Error(
|
||
"coalesce config write failed", "path", store.CoalescePath(), "error", err)
|
||
writeError(w, apperr.Wrap(apperr.ErrInternal, "coalesce_write_failed",
|
||
"面板配置文件落盘失败（盘上内容未改动）：%v", err))
|
||
return
|
||
}
|
||
}
|
||
log := logger.WithSource(s.log, coalesceLogSource)
|
||
if wrote {
|
||
log.Info("coalesce config saved (hot: the bridge rereads it before every departure)",
|
||
"path", store.CoalescePath(), "action", coalesceAuditAction)
|
||
} else {
|
||
log.Info("coalesce config unchanged (no bytes written)", "path", store.CoalescePath())
|
||
}
|
||
s.auditClaims(r, coalesceAuditAction, "ok")
|
||
writeJSON(w, http.StatusOK, buildCoalesceView(store))
|
||
}
|
||
|
||
// coalesceStore 解析「渠道静默窗口配置」的可选能力面。
|
||
func (s *Server) coalesceStore() (CoalesceConfigStore, bool) {
|
||
if s.channels == nil {
|
||
return nil, false
|
||
}
|
||
store, ok := s.channels.(CoalesceConfigStore)
|
||
if !ok || store.CoalescePath() == "" {
|
||
return nil, false
|
||
}
|
||
return store, true
|
||
}
|
||
|
||
// parseCoalescePatch 解析 PUT 请求体：严格解码（未知键即错），并做「敲错级」校验
|
||
// （负数）。越界的合法数字不在这里拒绝 —— 它们由运行期 clamp 并留 WARN。
|
||
func parseCoalescePatch(body []byte) (runtimeconfig.CoalescePatch, error) {
|
||
var patch runtimeconfig.CoalescePatch
|
||
dec := json.NewDecoder(bytes.NewReader(body))
|
||
dec.DisallowUnknownFields()
|
||
if err := dec.Decode(&patch); err != nil {
|
||
return runtimeconfig.CoalescePatch{}, fmt.Errorf("请求体不是合法的窗口参数补丁：%w", err)
|
||
}
|
||
if dec.More() {
|
||
return runtimeconfig.CoalescePatch{}, fmt.Errorf("请求体只允许一个 JSON 对象")
|
||
}
|
||
if patch.Default != nil {
|
||
if err := runtimeconfig.ValidateCoalescePatch(*patch.Default); err != nil {
|
||
return runtimeconfig.CoalescePatch{}, err
|
||
}
|
||
}
|
||
for key, conv := range patch.Conversations {
|
||
if conv == nil {
|
||
continue
|
||
}
|
||
if err := runtimeconfig.ValidateCoalescePatch(*conv); err != nil {
|
||
return runtimeconfig.CoalescePatch{}, fmt.Errorf("%s：%w", key, err)
|
||
}
|
||
}
|
||
return patch, nil
|
||
}
|
||
|
||
// buildCoalesceView 组装面板视图：盘上原值 + 生效值 + 取值规则 + 只读事实。
|
||
func buildCoalesceView(store CoalesceConfigStore) AdminCoalesceView {
|
||
file := store.CoalesceSnapshot()
|
||
view := AdminCoalesceView{
|
||
Path: store.CoalescePath(),
|
||
Present: store.CoalescePresent(),
|
||
Hot: true,
|
||
Default: adminCoalescePolicy(file.EffectiveDefault()),
|
||
Patch: file.Default.Values(),
|
||
Conversations: map[string]AdminCoalescePolicy{},
|
||
Rules: AdminCoalesceRules{
|
||
WindowMs: [2]int{runtimeconfig.CoalesceWindowMinMs, runtimeconfig.CoalesceWindowMaxMs},
|
||
MaxWaitMs: [2]int{runtimeconfig.CoalesceWindowMinMs, runtimeconfig.CoalesceMaxWaitMaxMs},
|
||
MaxBatchMessages: [2]int{runtimeconfig.CoalesceBatchMin, runtimeconfig.CoalesceBatchMax},
|
||
Clamp: "clamp",
|
||
FlushReasons: []string{"quiet", "max_wait", "all_stale", "off", "stop"},
|
||
},
|
||
OwnerPrivate: coalesceOwnerPrivateFact,
|
||
}
|
||
for key := range file.Conversations {
|
||
if policy, ok := file.EffectiveConversation(key); ok {
|
||
view.Conversations[key] = adminCoalescePolicy(policy)
|
||
}
|
||
}
|
||
return view
|
||
}
|
||
|
||
// adminCoalescePolicy 把 runtimeconfig 的策略投影成面板形状（字段一一对应）。
|
||
func adminCoalescePolicy(p runtimeconfig.CoalescePolicy) AdminCoalescePolicy {
|
||
return AdminCoalescePolicy{
|
||
Enabled: p.Enabled,
|
||
WindowMs: p.WindowMs,
|
||
MaxWaitMs: p.MaxWaitMs,
|
||
MaxBatchMessages: p.MaxBatchMessages,
|
||
ApplyToStrangers: p.ApplyToStrangers,
|
||
StrangerWindowMs: p.StrangerWindowMs,
|
||
}
|
||
}
|