feat(gateway): 入库队列接线 + 正文一律MinIO + file_id级联删端点

- 入库改走 JetStream 队列：claim-check 暂存 MinIO → 发作业 → 立即返 job_id； worker 池(StartIngestWorkers)有界并发消费，崩溃重投续跑(幂等)，优雅 drain - 存储：正文一律落 MinIO(去 <8000字内联PG 阈值，仅 MinIO 挂时回退兜底)； sundynix_doc 删死字段 MD5 - 下游键传稳定 file_id(非展示名)；新增 DELETE /api/v1/kb/doc 级联删 (三库 + MinIO 原文 + PG 元数据/双链，owner 隔离) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 13:38:03 +08:00
parent 79e834e8e9
commit 9a5ed75c05
6 changed files with 197 additions and 37 deletions
@@ -2,7 +2,6 @@ package handler

 import (
 	"context"
-	"crypto/md5"
 	"crypto/rand"
 	"encoding/base64"
 	"encoding/hex"
@@ -19,12 +18,12 @@ import (

 	"github.com/gin-gonic/gin"

+	"github.com/sundynix/sundynix-gateway/internal/blob"
+	"github.com/sundynix/sundynix-gateway/internal/nats"
+	"github.com/sundynix/sundynix-gateway/internal/store"
 	"github.com/sundynix/sundynix-shared/contract"
 )

-// docInlineMax 是内联存 PG 的正文字数上限；超过则正文落 MinIO，PG 只留元数据+预览+对象键。
-const docInlineMax = 8000
-
 // rawKB 规整知识库名（去空白，空则 default）—— 注册表里的展示名。
 func rawKB(kb string) string {
 	kb = strings.TrimSpace(kb)
@@ -82,8 +81,11 @@ func (h *Handler) KbIngest(c *gin.Context) {
 		return
 	}
 	_ = h.db.EnsureKB(c.Request.Context(), userID(c), rawKB(body.KB), "general")
-	job := newJobID()
-	go h.runIngest(job, userID(c), rawKB(body.KB), scopedKB(c, body.KB), "", "", nil, body.Text)
+	job, err := h.enqueueIngest(c.Request.Context(), userID(c), rawKB(body.KB), scopedKB(c, body.KB), "", "", nil, body.Text)
+	if err != nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()})
+		return
+	}
 	c.JSON(http.StatusAccepted, gin.H{"job_id": job})
 }

@@ -101,9 +103,12 @@ func (h *Handler) KbSaveNote(c *gin.Context) {
 	}
 	owner := userID(c)
 	_ = h.db.EnsureKB(c.Request.Context(), owner, rawKB(body.KB), "general")
-	// 落库 + 重建索引由后台 runIngest 统一处理（forceDoc=name 保持笔记身份）。
-	job := newJobID()
-	go h.runIngest(job, owner, rawKB(body.KB), scopedKB(c, body.KB), body.Name, "", nil, body.Content)
+	// 落库 + 重建索引由入库工作队列统一处理（forceDoc=name 保持笔记身份）。
+	job, err := h.enqueueIngest(c.Request.Context(), owner, rawKB(body.KB), scopedKB(c, body.KB), body.Name, "", nil, body.Content)
+	if err != nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()})
+		return
+	}
 	c.JSON(http.StatusAccepted, gin.H{"job_id": job, "name": body.Name})
 }

@@ -153,6 +158,39 @@ func (h *Handler) KbDoc(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{"id": d.ID, "name": d.Name, "ext": d.Ext, "content": content, "size": d.Size})
 }

+// KbDeleteDoc: DELETE /api/v1/kb/doc?id= —— 级联删一份文档：
+// 三库(向量/全文/图谱，经 mcp-go kb_delete 按 file_id) + MinIO 原文 + PG 元数据/双链。owner 作用域防越权。
+func (h *Handler) KbDeleteDoc(c *gin.Context) {
+	owner := userID(c)
+	d, err := h.db.GetDocByID(c.Request.Context(), owner, c.Query("id"))
+	if err != nil || d == nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "文档不存在"})
+		return
+	}
+	scoped := d.Owner + "/" + d.KB
+	// ① 三库按 file_id 级联删（失败不中断，避免半残；记录降级）。
+	if res, e := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("kb_delete"),
+		&contract.ToolCall{Tool: "kb_delete", Args: map[string]any{"kb": scoped, "file_id": d.ID}}); e != nil || res == nil || !res.OK {
+		msg := "kb_delete 无响应"
+		if e != nil {
+			msg = e.Error()
+		} else if res != nil {
+			msg = res.Error
+		}
+		log.Printf("[gateway] 三库删除降级 id=%s: %s", d.ID, msg)
+	}
+	// ② MinIO 原文。
+	if d.ObjectKey != "" && h.blob.Ready() {
+		h.blob.Delete(c.Request.Context(), d.ObjectKey)
+	}
+	// ③ PG 元数据 + 双链。
+	if e := h.db.DeleteDocByID(c.Request.Context(), owner, d.KB, d.ID); e != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": e.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"deleted": d.ID, "name": d.Name})
+}
+
 // KbLinks: GET /api/v1/kb/links?kb= —— 某库已解析的 [[双链]]（FromID→ToID），供反链/笔记关系图按 ID 渲染。
 func (h *Handler) KbLinks(c *gin.Context) {
 	rows, err := h.db.ListLinks(c.Request.Context(), userID(c), rawKB(c.Query("kb")))
@@ -203,28 +241,91 @@ func (h *Handler) KbIngestFile(c *gin.Context) {
 		return
 	}
 	_ = h.db.EnsureKB(c.Request.Context(), userID(c), rawKB(kb), "general")
-	job := newJobID()
-	go h.runIngest(job, userID(c), rawKB(kb), scopedKB(c, kb), "", fh.Filename, data, "")
+	job, err := h.enqueueIngest(c.Request.Context(), userID(c), rawKB(kb), scopedKB(c, kb), "", fh.Filename, data, "")
+	if err != nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()})
+		return
+	}
 	c.JSON(http.StatusAccepted, gin.H{"job_id": job, "file": fh.Filename})
 }

-// runIngest 后台跑入库流水线，逐阶段把进度发到 sundynix.streams.<job>。
+// StartIngestWorkers 启动入库 worker 池：在 JetStream 入库队列上有界并发消费作业（背压、崩溃重投）。
+// 返回 drain 供优雅停机时等在途入库跑完（超时未完的由 AckWait 在重启后重投兜底）。
+func StartIngestWorkers(ctx context.Context, db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.Store) (func(context.Context), error) {
+	h := New(db, cache, bus, blobStore)
+	return bus.ConsumeIngestJobs(ctx, h.processIngestJob)
+}
+
+// enqueueIngest 把一次入库请求暂存到对象存储(claim-check)，再发布作业到 JetStream 工作队列，
+// 立即返回 job_id。暂存让作业消息恒小（不受 max_payload 限制）、崩溃重投只搬引用。
+// MinIO 是准生产硬依赖（正文也一律落 MinIO）；未就绪即拒绝入库，不静默退化。
+func (h *Handler) enqueueIngest(ctx context.Context, owner, kbName, scoped, forceDoc, filename string, data []byte, rawText string) (string, error) {
+	if !h.blob.Ready() {
+		return "", errors.New("对象存储未就绪，暂时无法入库")
+	}
+	job := newJobID()
+	stageKey := "ingest-staging/" + job
+	isText := filename == ""
+	payload := rawText
+	if !isText {
+		payload = string(data) // 原始文件字节（string 可承载任意字节）
+	}
+	if err := h.blob.Put(ctx, stageKey, payload); err != nil {
+		return "", fmt.Errorf("暂存失败: %w", err)
+	}
+	jobMsg := &contract.IngestJob{
+		JobID: job, Owner: owner, KBName: kbName, Scoped: scoped,
+		ForceDoc: forceDoc, Filename: filename, StageKey: stageKey, IsText: isText,
+	}
+	if err := h.bus.PublishIngestJob(ctx, jobMsg); err != nil {
+		h.blob.Delete(ctx, stageKey) // 入队失败 → 清暂存避免泄漏
+		return "", fmt.Errorf("入队失败: %w", err)
+	}
+	return job, nil
+}
+
+// processIngestJob 是入库 worker：从暂存取回原料 → 跑入库流水线 → 成功/彻底失败后清暂存。
+// 返回非 nil → 队列延迟重投（仅瞬时基建失败且未到最后一次投递时）。幂等：按 doc 先删后写，重跑安全。
+func (h *Handler) processIngestJob(ctx context.Context, job *contract.IngestJob, lastAttempt bool) error {
+	payload, err := h.blob.Get(ctx, job.StageKey)
+	if err != nil {
+		// 暂存读不到（已清/对象丢失）——无法恢复，别无限重试，直接终态收尾。
+		_ = h.bus.PublishIngest(job.JobID, &contract.IngestEvent{Stage: "失败", Error: "暂存读取失败: " + err.Error()})
+		_ = h.bus.CompleteStream(job.JobID)
+		return nil
+	}
+	var data []byte
+	var rawText string
+	if job.IsText {
+		rawText = payload
+	} else {
+		data = []byte(payload)
+	}
+	retryable, rerr := h.runIngest(ctx, job.JobID, job.Owner, job.KBName, job.Scoped, job.ForceDoc, job.Filename, data, rawText)
+	if rerr != nil && retryable && !lastAttempt {
+		return rerr // 瞬时失败 → 保留暂存，延迟重投
+	}
+	h.blob.Delete(context.Background(), job.StageKey) // 成功或彻底失败 → 清暂存
+	return nil
+}
+
+// runIngest 跑入库流水线，逐阶段把进度发到 sundynix.streams.<job>，由入库 worker 调用。
 // owner+kbName 用于"文库"原文留存；scoped 是 owner/kb 作向量/全文/图谱分区键。
 // forceDoc 非空时强制以它为文档名（笔记编辑用，保持笔记身份稳定）。
 // filename 非空表示文件入库（先经 mcp-py 解析）；否则用 rawText。
-func (h *Handler) runIngest(job, owner, kbName, scoped, forceDoc, filename string, data []byte, rawText string) {
-	ctx := context.Background()
+// 返回 (retryable, err)：解析失败=终态(坏输入重试无益)；kb_ingest 基建失败=瞬时可重试。
+func (h *Handler) runIngest(ctx context.Context, job, owner, kbName, scoped, forceDoc, filename string, data []byte, rawText string) (retryable bool, err error) {
 	emit := func(ev contract.IngestEvent) { _ = h.bus.PublishIngest(job, &ev) }
 	time.Sleep(400 * time.Millisecond) // 给 SSE 客户端订阅时间（core NATS 无缓冲）

 	text := rawText
 	if filename != "" {
 		emit(contract.IngestEvent{Stage: "解析", Msg: filename})
-		parsed, err := h.parseFile(ctx, filename, data)
-		if err != nil {
-			emit(contract.IngestEvent{Stage: "失败", Error: "解析失败: " + err.Error()})
+		parsed, perr := h.parseFile(ctx, filename, data)
+		if perr != nil {
+			emit(contract.IngestEvent{Stage: "失败", Error: "解析失败: " + perr.Error()})
 			_ = h.bus.CompleteStream(job)
-			return
+			return false, fmt.Errorf("解析失败: %w", perr) // 终态：坏输入
 		}
 		emit(contract.IngestEvent{
 			Stage:   "解析完成",
@@ -242,25 +343,28 @@ func (h *Handler) runIngest(job, owner, kbName, scoped, forceDoc, filename strin
 	if docName == "" {
 		docName = noteName(text)
 	}
+	var fileID string // 文档稳定 ID（雪花），作 Milvus/Bleve/Neo4j 的关联键——重名/重入库不变，删可级联
 	if text != "" {
 		size := len([]rune(text))
 		ext := strings.ToLower(filepath.Ext(filename)) // 笔记/文本入库时 filename 为空 → ext 为空
-		sum := md5.Sum([]byte(text))
-		md5hex := hex.EncodeToString(sum[:])
 		inline, objectKey := text, ""
-		// 大文档正文落对象存储，PG 只留元数据+预览+对象键（避免把十几万字塞进 PG）。
-		if size > docInlineMax && h.blob.Ready() {
+		// 正文一律落对象存储（MinIO），PG 只留元数据+预览+对象键（不分大小，不把正文塞进 PG）。
+		// 仅当 MinIO 不可用或写失败时，才回退内联，保证正文不丢。
+		if h.blob.Ready() {
 			key := owner + "/" + kbName + "/" + docName
 			if err := h.blob.Put(ctx, key, text); err == nil {
 				inline, objectKey = "", key
 			} else {
-				log.Printf("[gateway] 大文档转 MinIO 失败，回退内联: %v", err)
+				log.Printf("[gateway] 正文转 MinIO 失败，回退内联: %v", err)
 			}
+		} else {
+			log.Printf("[gateway] MinIO 未就绪，正文回退内联存 PG（doc=%s）", docName)
 		}
-		docID, oldKey, err := h.db.SaveDoc(ctx, owner, kbName, docName, ext, md5hex, inline, objectKey, size, head(text, 500))
+		docID, oldKey, err := h.db.SaveDoc(ctx, owner, kbName, docName, ext, inline, objectKey, size, head(text, 500))
 		if err != nil {
 			log.Printf("[gateway] 文件入库失败: %v", err)
 		} else if docID != "" {
+			fileID = docID // 下游三库用它作关联键
 			// 孤儿 GC：重名覆盖后旧对象键若已不用（转内联或换键），从 MinIO 删除，避免泄漏。
 			if oldKey != "" && oldKey != objectKey && h.blob.Ready() {
 				h.blob.Delete(ctx, oldKey)
@@ -271,19 +375,26 @@ func (h *Handler) runIngest(job, owner, kbName, scoped, forceDoc, filename strin
 		}
 	}

-	// 调 mcp-go kb_ingest（带 job_id）：它会发 切块/向量化/写入/完成 事件 + CompleteStream。
-	res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("kb_ingest"),
-		&contract.ToolCall{Tool: "kb_ingest", Args: map[string]any{"kb": scoped, "doc": docName, "text": text, "job_id": job}})
-	if err != nil || res == nil || !res.OK {
+	// 调 mcp-go kb_ingest：doc 传**稳定 file_id**（非展示名），作下游三库关联键，使重名/重入库幂等、删可级联。
+	// SaveDoc 失败兜底退回用名字做键（仍能入库，只是失去 file_id 稳定性）。
+	docKey := fileID
+	if docKey == "" {
+		docKey = docName
+	}
+	res, cerr := h.bus.CallTool(ctx, contract.ToolSubjectGo("kb_ingest"),
+		&contract.ToolCall{Tool: "kb_ingest", Args: map[string]any{"kb": scoped, "doc": docKey, "text": text, "job_id": job}})
+	if cerr != nil || res == nil || !res.OK {
 		msg := "kb_ingest 失败"
-		if err != nil {
-			msg = err.Error()
+		if cerr != nil {
+			msg = cerr.Error()
 		} else if res != nil {
 			msg = res.Error
 		}
 		emit(contract.IngestEvent{Stage: "失败", Error: msg})
 		_ = h.bus.CompleteStream(job)
+		return true, errors.New(msg) // 瞬时：mcp-go/embedding/Milvus 抖动，可重试
 	}
+	return false, nil
 }

 // KbIngestStream: GET /api/v1/kb/ingest/:id/stream —— SSE 实时推送入库进度事件。