refactor+feat(documents,channelbridge): 共享存储核心 + IM 入站附件摄取(主人 2.3)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 抽共享存储核心:internal/runtime/documents/store.go(22 项白名单 + 字节嗅探唯一裁决 + 内容寻址 sha256[:32]+ext + 去重 + 限额 + 路径双检)+ store_meta.go(documents 表唯一写入实现)。

- gateway 改为调用 store,**对外行为零变化**(机器证明:既有 documents_test/chat_documents_test/chat_documents_bounds_test 三个文件 0 行改动且全绿;状态码/响应字段/错误码/审计动作名/限额文档口径全部逐字保留;maxChatDocuments 改为引用 documents.MaxPerMessage)。

- IM 入站附件摄取(channelbridge):文件类媒体 → store.Save → 挂 model.DocumentMessage(RoleUser) 部件 ⇒ 后续 resolveDocuments 自动抽取/注入/清单;挂载点在「会话解析 + 黑白名单 + 最小打扰」三重闸门之后(被拦消息绝不在盘上留文件);**不支持的格式不静默**(诚实提示行 + 带来源日志,只记 name/mime/size,名字做展示净化防伪造行);图片/语音保持既有链路。

- 主会话独立验收:隔离 worktree(M1 的 12 文件)build/vet 绿、全树 59 包 0 FAIL、双平台交叉编译绿、12 条 Inbound 用例 PASS;变异独立复现(关闭 IM 摄取 ⇒ 5 条红,含「落盘文件数 = [] want 1」)。

- ⚠️ 真机边界(需裁决):现网四个适配器上报的文件媒体**只带 URL**(telegram FileID / qq-obv11 file|url / weixin·kook CDN URL),Media.Data 恒空 ⇒ 本切片只摄**内联字节**,URL 一律「诚实提示,不代取远程」(避免第三份 SSRF 实现)。真机要让「她读到 IM 文件」,需各适配器入站时下载并内联(凭据在各自渠道里,推荐)或新增共享受守卫抓取器。

- 另登记:IM 附件暂未写 chat_messages.document 列(二手回合看不到引用,建议并入 S7);channelbridge 既有测试夹具在 -race 下有数据竞争(stubSessions.GetOrCreate,与本次无关)。
This commit is contained in:
AskaEth committed 2026-10-04 19:25:00 +08:00
1 parent 51c1672355
commit 2a43035f5d
12 files changed
+1743 -158

No files matched your search

+4 -1
View File
@@ -22,11 +22,14 @@ import (
"strings"
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// maxChatDocuments 是**单回合**附件数上限(design/24 §6.1、§9.4-3 的定稿值 5)。
const maxChatDocuments = 5
// 取值取自共享存储核心(切片 M1):Web 上传的单回合上限与 IM 入站附件的单条
// 上限因此同源,不可能一处改宽、另一处照旧。
const maxChatDocuments = documents.MaxPerMessage
// tooManyDocumentsCode 是附件数超限的机器可读错误码(design/24 §6.2 表格)。
// 它与 image 侧的 image_too_large 同一风格:客户端只看 code 就能分开处理。
+85 -154
View File
@@ -16,9 +16,6 @@ package gateway
// 元数据表 `documents` 只写不回收(引用计数与 GC 属切片 S7)。
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
@@ -31,17 +28,19 @@ import (
"strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
)
const (
// maxDocumentBytes 是 `POST /api/v1/documents` 的**单文件字节**默认上限:20 MiB。
// maxDocumentBytes 是 `POST /api/v1/documents` 的**单文件字节**默认上限:
// 它就是共享存储核心的 `documents.DefaultMaxBytes`(20 MiB)—— 全仓只定义
// 一次,网关与 IM 入站附件因此不可能各写一个数(design/24 §3.3)。
// 它与图片上限(`maxUploadBytes` / `CHAT_IMAGE_MAX_BYTES`)**刻意解耦**:
// 图片上限被 vision 解码层钳住,文档上限由本项目的解析与上下文预算决定;
// 共用同一个数会让「改文档上限」静默改掉图片行为(design/24 §3.3)。
maxDocumentBytes = 20 << 20
maxDocumentBytes = documents.DefaultMaxBytes
// documentMultipartSlack 是 multipart 信封(边界、部件头)在文件上限之上
// 预留的字节余量;文件本身的校验按确切上限做。
@@ -53,7 +52,8 @@ const (
// documentHashHexLen 是内容寻址落盘名使用的 sha256 十六进制前缀长度
// (32 位十六进制 = 128 位),与 `validUploadName` 的形态校验同宽。
documentHashHexLen = 32
// 它取自共享存储核心,与落盘/读侧的名字形态同源(切片 M1)。
documentHashHexLen = documents.HashHexLen
// documentFormField 是上传 multipart 里的文件字段名(design/24 §6.1)。
documentFormField = "file"
@@ -75,8 +75,8 @@ const (
// documentSniffOKMIME 是唯一放行的嗅探结果基类型。白名单里的每一种文档
// (文本、代码、CSV、JSON、YAML…)经 `http.DetectContentType` 嗅探后都落在
// `text/plain`;HTML/XML/SVG/压缩包/可执行文件/判不定者各有自己的签名,
// 因此「不是 text/plain」即「拒绝」。
documentSniffOKMIME = "text/plain"
// 因此「不是 text/plain」即「拒绝」。取值取自共享存储核心(同一裁决)。
documentSniffOKMIME = documents.SniffOKMIME
// documentFallbackMIME 是读侧嗅探不出放行类型时下发的媒体类型。配合
// attachment + nosniff,它保证浏览器永远不会按类型渲染文档。
@@ -91,10 +91,6 @@ const (
documentUploadFailedCode = "document_upload_failed"
)
// ErrDocumentTooLarge 报告文档体量超过生效上限。`storeDocument` 用它把
// 「超限(413)」与「存储失败(500)」分开,调用方用 `errors.Is` 判定。
var ErrDocumentTooLarge = errors.New("gateway: document too large")
// documentResponse 是上传成功的载荷。URL 是**相对**同源路径(design/24 §6.1):
// 读侧要求 Bearer,因此该 URL 不是「贴进 `<img>` 就能取」的能力 URL,绝对化
// 没有意义;客户端真正要用的是 `chat.documents` 里的 `name`。
@@ -107,23 +103,6 @@ type documentResponse struct {
Deduped bool `json:"deduped"`
}
// documentRecord 是写入 `documents` 表的一行元数据(S1 只写)。
type documentRecord struct {
Name string
SHA256 string
MIME string
Size int64
Uploader string
}
// documentStored 是一次成功落盘的结果。
type documentStored struct {
name string
sha256 string
size int64
deduped bool
}
// documentLimitBytes 解析本次生效的单文件字节上限:配置给了正数就用它,
// 否则(未配置的零值、0、负数)回落到默认 20 MiB —— 语义是「≤0 = 用默认」
// 而不是「禁用上限」,与图片侧 `chatImageLimitBytes` 的口径一致。
@@ -138,23 +117,18 @@ func documentLimitBytes(configured int) int {
// documentExtOf 从客户端原始文件名取出小写扩展名(含点)。它只作参考:
// 结果必须同时通过扩展名白名单与字节嗅探,客户端因此无法靠改扩展名换类型。
// 取值逻辑与共享存储核心同源(切片 M1),读侧不再各写一套。
func documentExtOf(filename string) string {
return strings.ToLower(filepath.Ext(filename))
return documents.ExtOf(filename)
}
// documentExtAllowed 报告扩展名是否在放行集合内。deny-by-default:集合之外的
// 一切(`.html`/`.svg`/`.xml`/`.zip`/`.docx`/`.xlsx`/`.exe`/`.dll`/`.bat`/
// `.cmd`/`.ps1`/`.js`/`.mjs`/`.sh`/`.py`…)一律拒绝。注意 `.js`/`.py` 即便
// 嗅探成 `text/plain` 也必须拒绝 —— 判定以扩展名与嗅探的**合取**为准。
// 白名单本体只存在于 `documents.ExtAllowed`(22 项,design/24 §9.2-2 定稿)。
func documentExtAllowed(ext string) bool {
switch ext {
case ".txt", ".md", ".markdown", ".csv", ".tsv", ".json", ".jsonl",
".yaml", ".yml", ".log", ".ini", ".cfg", ".toml",
".go", ".ts", ".tsx", ".java", ".c", ".h", ".cpp", ".rs", ".sql":
return true
default:
return false
}
return documents.ExtAllowed(ext)
}
// documentBaseMIME 把嗅探结果归一为不含参数的基类型(`text/plain; charset=utf-8`
@@ -169,14 +143,9 @@ func documentBaseMIME(sniffed string) string {
// documentSniffAllowed 报告嗅探所得类型是否落在放行的文本类内。它是「类型裁决
// 的唯一裁决者」在代码里的落点:客户端声明的 `Content-Type` 从不参与决定。
// 裁决本体只存在于 `documents.SniffAllowed`(写侧与读侧同一份)。
func documentSniffAllowed(sniffed string) bool {
return documentBaseMIME(sniffed) == documentSniffOKMIME
}
// documentHashName 返回内容寻址的落盘名:sha256 十六进制的前 32 位 + ext。
// 同一份字节配同一个扩展名必然得到同一个名字,去重因此是「同名」的自然结果。
func documentHashName(sum []byte, ext string) string {
return hex.EncodeToString(sum)[:documentHashHexLen] + ext
return documents.SniffAllowed(sniffed)
}
// validDocumentName 报告 name 是否恰好是 32 位小写十六进制主名加白名单文档
@@ -238,10 +207,45 @@ func (s *Server) readDocumentUpload(w http.ResponseWriter, r *http.Request, limi
return file, header, true
}
// documentStore 按本次生效的目录与上限构造一个文档存储(切片 M1 抽出的共享
// 核心,见 `internal/runtime/documents`):网关只是它的一个调用方 —— IM 入站
// 附件走的是**同一个** Store,两条来源因此不可能漂移。上限来自
// `CHAT_DOCUMENT_MAX_BYTES`(经 documentLimitBytes 归一),并且**每个请求**
// 按当时的配置取值,面板改配置后下一个请求即生效。
//
// 元数据写入面由 `MetadataWriterFor` 给出:未接库(`s.dbPath` 为空)时它返回
// 真正的 nil 接口,Store 因此按「未注入」跳过写表并留痕(S1 的行为)。
func (s *Server) documentStore(dir string, limit int, uploader string) *documents.Store {
return documents.NewStore(documents.StoreConfig{
UploadsDir: dir,
MaxBytes: limit,
Uploader: uploader,
Meta: documents.MetadataWriterFor(s.dbPath, s.log),
Log: s.log,
})
}
// readDocumentBytes 读出本次上传的文件字节,最多 `limit+1` 字节:多读一个字节
// 就足以判定超限,因此既不把超限内容放大进内存,也不把它留在盘上。S1 的流式
// 落盘在抽成共享存储后改为「先读进内存再交给 store」——上限本身(默认 20 MiB)
// 没有放宽,而文档抽取路径本来就整份读进内存(design/24 §9.6 已登记的同一代价)。
func readDocumentBytes(file io.Reader, limit int) ([]byte, error) {
data, err := io.ReadAll(io.LimitReader(file, int64(limit)+1))
if err != nil {
return nil, fmt.Errorf("documents: 读上传内容: %w", err)
}
return data, nil
}
// handleDocumentUpload 接受带单个 `file` 部件的 multipart/form-data 请求,
// 用字节嗅探裁决类型,按内容寻址存到 `<dataDir>/uploads/documents`,写一行
// 元数据并审计 `document.upload`。首次上传回 201 + `deduped:false`;命中同一
// 内容时回 200 + `deduped:true` 且 `name` 与首次逐字一致。
//
// 类型/体积的裁决**全部**由 `documents.Store.Save` 给出(切片 M1 起与 IM 入站
// 附件同源);这里只把它的哨兵错误映射为 S1 既有的契约错误码,判定顺序
// (扩展名 → 空 → 上限 → 嗅探)由 store 保持,因此线上答案与抽共享核心之前
// 逐字一致。
func (s *Server) handleDocumentUpload(w http.ResponseWriter, r *http.Request) {
limit := documentLimitBytes(s.cfg.Server.ChatDocumentMaxBytes)
file, header, ok := s.readDocumentUpload(w, r, limit)
@@ -250,139 +254,66 @@ func (s *Server) handleDocumentUpload(w http.ResponseWriter, r *http.Request) {
}
defer func() { _ = file.Close() }()
ext := documentExtOf(header.Filename)
if !documentExtAllowed(ext) {
// 扩展名先于读体判定:明显不该收的扩展名不必先把请求体读进内存(S1 的
// 顺位也是先看扩展名)。判定用的是**共享存储核心的同一份白名单**,
// 与 store 内部的裁决不可能分叉;store 仍会再判一次(防御性)。
if !documents.ExtAllowed(documents.ExtOf(header.Filename)) {
s.failDocumentUpload(w, r, documentUnsupportedError())
return
}
if header.Size <= 0 {
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "文件为空"))
return
}
if header.Size > int64(limit) {
s.failDocumentUpload(w, r, documentTooLargeError(limit))
return
}
sniffed, err := sniffImageMIME(file)
data, err := readDocumentBytes(file, limit)
if err != nil {
s.documentsLog().Error("document sniff", "err", err)
s.documentsLog().Error("document read", "err", err)
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "读取上传内容失败"))
return
}
if !documentSniffAllowed(sniffed) {
s.failDocumentUpload(w, r, documentUnsupportedError())
return
}
dir, err := s.documentsDir()
if err != nil {
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "文档存储目录未配置"))
return
}
stored, err := s.storeDocument(dir, ext, file, int64(limit))
if errors.Is(err, ErrDocumentTooLarge) {
s.failDocumentUpload(w, r, documentTooLargeError(limit))
return
}
ref, err := s.documentStore(dir, limit, documentUploader(r)).Save(r.Context(), header.Filename, data)
if err != nil {
s.documentsLog().Error("document store", "err", err)
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "保存文档失败"))
return
}
rec := documentRecord{
Name: stored.name,
SHA256: stored.sha256,
MIME: documentBaseMIME(sniffed),
Size: stored.size,
Uploader: documentUploader(r),
}
if err := s.insertDocumentMetadata(r.Context(), rec); err != nil {
s.documentsLog().Error("document metadata", "err", err, "name", rec.Name)
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "记录文档元数据失败"))
s.failDocumentSave(w, r, limit, ref, err)
return
}
s.documentsLog().Info("document stored",
"name", rec.Name, "size", rec.Size, "mime", rec.MIME,
"sha256", rec.SHA256, "deduped", stored.deduped)
"name", ref.Name, "size", ref.Size, "mime", ref.MIME,
"sha256", ref.SHA256, "deduped", ref.Deduped)
s.auditClaims(r, documentUploadAction, "ok")
status := http.StatusCreated
if stored.deduped {
if ref.Deduped {
status = http.StatusOK
}
writeJSON(w, status, documentResponse{
Name: rec.Name,
URL: documentFilePrefix + rec.Name,
MIME: rec.MIME,
Size: rec.Size,
SHA256: rec.SHA256,
Deduped: stored.deduped,
Name: ref.Name,
URL: documentFilePrefix + ref.Name,
MIME: ref.MIME,
Size: ref.Size,
SHA256: ref.SHA256,
Deduped: ref.Deduped,
})
}
// storeDocument 把 f 按内容寻址写入 dir:先流式写点前缀临时文件、同时计算
// sha256,再把临时文件重命名到 `<sha256[:32]><ext>`(部分写入因此绝不会以可读的
// 名字露面)。目标名已存在(同字节同扩展名)时丢弃临时文件并报告 deduped,
// 盘上只留一份。目录 0700、文件继承临时的 0600。
func (s *Server) storeDocument(dir, ext string, f io.Reader, limit int64) (documentStored, error) {
if err := os.MkdirAll(dir, 0o700); err != nil {
return documentStored{}, fmt.Errorf("create document dir: %w", err)
// failDocumentSave 把 store 的哨兵错误映射为 S1 既有的线上错误并审计。映射表
// 逐条对应(415 类型拒绝 / 413 超限 / 400 空文件 / 500 存储或元数据失败),
// 与抽共享核心之前的对外行为一字不差:拒绝是**可预期的输入问题**,因此不额外
// 记错误日志(与 S1 相同);存储与元数据失败才留日志。
func (s *Server) failDocumentSave(w http.ResponseWriter, r *http.Request, limit int, ref documents.Ref, err error) {
switch {
case errors.Is(err, documents.ErrTooLarge):
s.failDocumentUpload(w, r, documentTooLargeError(limit))
case errors.Is(err, documents.ErrExtNotAllowed), errors.Is(err, documents.ErrContentNotText):
s.failDocumentUpload(w, r, documentUnsupportedError())
case errors.Is(err, documents.ErrEmpty):
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "文件为空"))
case errors.Is(err, documents.ErrMetadata):
s.documentsLog().Error("document metadata", "err", err, "name", ref.Name)
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "记录文档元数据失败"))
default:
s.documentsLog().Error("document store", "err", err)
s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "保存文档失败"))
}
tmp, err := os.CreateTemp(dir, ".document-*")
if err != nil {
return documentStored{}, fmt.Errorf("create document temp: %w", err)
}
tmpName := tmp.Name()
hasher := sha256.New()
written, err := io.Copy(io.MultiWriter(tmp, hasher), io.LimitReader(f, limit+1))
if err != nil {
_ = tmp.Close()
_ = os.Remove(tmpName)
return documentStored{}, fmt.Errorf("write document: %w", err)
}
if err := tmp.Close(); err != nil {
_ = os.Remove(tmpName)
return documentStored{}, fmt.Errorf("close document: %w", err)
}
// 文件级上限的第二次把关(第一道是 multipart 部件大小):多读一个字节即可
// 判定超限,因此既不放大内存也不把超限文件留在盘上。
if written > limit {
_ = os.Remove(tmpName)
return documentStored{}, ErrDocumentTooLarge
}
digest := hasher.Sum(nil)
sum := hex.EncodeToString(digest)
name := documentHashName(digest, ext)
if info, serr := os.Stat(filepath.Join(dir, name)); serr == nil && !info.IsDir() {
_ = os.Remove(tmpName)
return documentStored{name: name, sha256: sum, size: written, deduped: true}, nil
}
if err := os.Rename(tmpName, filepath.Join(dir, name)); err != nil {
_ = os.Remove(tmpName)
return documentStored{}, fmt.Errorf("commit document: %w", err)
}
return documentStored{name: name, sha256: sum, size: written, deduped: false}, nil
}
// insertDocumentMetadata 把一行文档元数据写进 `documents` 表(迁移 0014)。
// 同一内容寻址名重复上传时 `DO NOTHING`:首次的 uploader/created_at 是权威,
// 去重不刷新它们。`s.dbPath` 未注入(未接数据库的构建/测试)时跳过 —— 上传
// 本身仍然成功,元数据留待 S7 的引用计数与 GC 统一接线。
func (s *Server) insertDocumentMetadata(ctx context.Context, rec documentRecord) error {
path := strings.TrimSpace(s.dbPath)
if path == "" {
return nil
}
conn, err := db.Open(ctx, path, s.log)
if err != nil {
return fmt.Errorf("open db for document metadata: %w", err)
}
defer func() { _ = conn.Close() }()
const query = `INSERT INTO documents (name, sha256, mime, size, uploader)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(name) DO NOTHING`
if _, err := conn.ExecContext(ctx, query, rec.Name, rec.SHA256, rec.MIME, rec.Size, rec.Uploader); err != nil {
return fmt.Errorf("insert document metadata: %w", err)
}
return nil
}
// documentUploader 从请求上下文取上传者 id(由 auth.Authorize 填充),取不到时