diff --git a/.gitignore b/.gitignore index bfd71f5..4c75354 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,7 @@ sundynix-desktop/frontend/package.json.md5 # Data data/ +.data/ # 演示产物 .bin/ diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index e07143e..7dc0b5b 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -13,6 +13,7 @@ import ( "github.com/sundynix/sundynix-gateway/internal/blob" "github.com/sundynix/sundynix-gateway/internal/nats" + "github.com/sundynix/sundynix-gateway/internal/handler" "github.com/sundynix/sundynix-gateway/internal/router" "github.com/sundynix/sundynix-gateway/internal/store" sharedbus "github.com/sundynix/sundynix-shared/bus" @@ -94,6 +95,14 @@ func main() { log.Printf("[gateway] subscribe usage: %v", err) } + // 入库工作队列：启动有界并发 worker 池消费 JetStream 入库作业（背压防 OOM、崩溃重投兜底）。 + ingestDrain, err := handler.StartIngestWorkers(context.Background(), db, cache, bus, blobStore) + if err != nil { + log.Printf("[gateway] 启动入库 worker 失败（入库不可用）: %v", err) + } else { + log.Printf("[gateway] 入库 worker 池就绪（JetStream 持久队列）") + } + r := router.New(db, cache, bus, blobStore) addr := envOr("GATEWAY_ADDR", ":8080") srv := &http.Server{Addr: addr, Handler: r} @@ -118,6 +127,13 @@ func main() { if err := srv.Shutdown(sctx); err != nil { log.Printf("[gateway] HTTP 优雅关闭超时/出错: %v", err) } + // HTTP 排空后再 drain 在途入库作业（≤to）；超时未完的不丢——AckWait 30min 内重启会重投续跑。 + if ingestDrain != nil { + log.Printf("[gateway] drain 在途入库作业 ≤%s（超时未完者重启后由队列重投）…", to) + ictx, icancel := context.WithTimeout(context.Background(), to) + ingestDrain(ictx) + icancel() + } log.Println("[gateway] 已优雅停机") } diff --git a/sundynix-gateway/internal/handler/kb.go b/sundynix-gateway/internal/handler/kb.go index e1d4455..85fbdbc 100644 --- a/sundynix-gateway/internal/handler/kb.go +++ b/sundynix-gateway/internal/handler/kb.go @@ -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.。 +// 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.，由入库 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 实时推送入库进度事件。 diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index 93e31ce..eaf0c9a 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -24,7 +24,10 @@ func MustConnect(url string) *Bus { if err := inner.EnsureTaskStream(context.Background()); err != nil { log.Fatalf("[nats] ensure stream: %v", err) } - log.Printf("[nats] connected %s, task stream ready", url) + if err := inner.EnsureIngestStream(context.Background()); err != nil { + log.Fatalf("[nats] ensure ingest stream: %v", err) + } + log.Printf("[nats] connected %s, task + ingest streams ready", url) return &Bus{inner: inner} } @@ -101,4 +104,19 @@ func (b *Bus) PublishIngest(jobID string, ev *contract.IngestEvent) error { // CompleteStream 发送入库流结束信号。 func (b *Bus) CompleteStream(jobID string) error { return b.inner.CompleteStream(jobID) } +// ---- 入库工作队列（JetStream 持久）---- + +// EnsureIngestStream 确保入库作业流存在（启动时调）。 +func (b *Bus) EnsureIngestStream(ctx context.Context) error { return b.inner.EnsureIngestStream(ctx) } + +// PublishIngestJob 把入库作业持久入队（崩溃不丢，由 worker 消费）。 +func (b *Bus) PublishIngestJob(ctx context.Context, job *contract.IngestJob) error { + return b.inner.PublishIngestJob(ctx, job) +} + +// ConsumeIngestJobs 启动入库 worker 池消费作业（有界并发 + 背压 + 崩溃重投）。 +func (b *Bus) ConsumeIngestJobs(ctx context.Context, h sharedbus.IngestHandler) (func(context.Context), error) { + return b.inner.ConsumeIngestJobs(ctx, h) +} + func (b *Bus) Close() { b.inner.Close() } diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index 8447ec2..fdd4e45 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -64,6 +64,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. p.POST("/kb/search", h.KbSearch) // 检索台 p.GET("/kb/vault", h.KbVault) // 文库列表 p.GET("/kb/doc", h.KbDoc) // 取单篇文档 + p.DELETE("/kb/doc", h.KbDeleteDoc) // 级联删文档（三库+MinIO+PG） p.GET("/kb/links", h.KbLinks) // 某库双链 p.POST("/kb/note", h.KbSaveNote) // 新建/编辑笔记 p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组 diff --git a/sundynix-gateway/internal/store/model.go b/sundynix-gateway/internal/store/model.go index f8bf280..7076397 100644 --- a/sundynix-gateway/internal/store/model.go +++ b/sundynix-gateway/internal/store/model.go @@ -95,11 +95,10 @@ type Doc struct { KB string `gorm:"size:64;uniqueIndex:idx_doc_okn"` Name string `gorm:"size:160;uniqueIndex:idx_doc_okn"` // 文件名称（不含扩展名 / 笔记名） Ext string `gorm:"size:16"` // 文件后缀（.md/.pdf/.docx…；笔记/文本为空） - MD5 string `gorm:"size:32;index"` // 正文内容指纹（去重 / 校验） Size int // 原文字数（rune） Preview string `gorm:"size:600"` // 前若干字预览（列表/反链用，不拉全文） - Content string `gorm:"type:text"` // 小文档内联正文；大文档转 MinIO 后置空 - ObjectKey string `gorm:"size:160"` // 存放链接：大文档在 MinIO 的对象键（空=内联 Content） + Content string `gorm:"type:text"` // 正文一律落 MinIO 后置空；仅 MinIO 不可用时回退内联存这里 + ObjectKey string `gorm:"size:160"` // 正文在 MinIO 的对象键（常态非空；空=回退内联 Content） } func (Doc) TableName() string { return "sundynix_doc" } @@ -120,14 +119,14 @@ func (DocLink) TableName() string { return "sundynix_doc_link" } // SaveDoc 写入/更新一份文件（owner+kb+name 唯一，重名覆盖），返回 (文件雪花 ID, 被覆盖的旧对象键)。 // oldObjectKey 是重名覆盖前该文档的 ObjectKey（新建则为空）—— 供调用方清理 MinIO 孤儿对象。 // content 为内联正文（大文档转 MinIO 时传空 + objectKey）；ext/md5/preview/size 由调用方按全文给出。 -func (p *Postgres) SaveDoc(ctx context.Context, owner, kb, name, ext, md5, content, objectKey string, size int, preview string) (id, oldObjectKey string, err error) { +func (p *Postgres) SaveDoc(ctx context.Context, owner, kb, name, ext, content, objectKey string, size int, preview string) (id, oldObjectKey string, err error) { if p.db == nil { return "", "", nil } var d Doc qerr := p.db.WithContext(ctx).Where("owner = ? AND kb = ? AND name = ?", owner, kb, name).First(&d).Error if errors.Is(qerr, gorm.ErrRecordNotFound) { - d = Doc{Owner: owner, KB: kb, Name: name, Ext: ext, MD5: md5, Content: content, ObjectKey: objectKey, Size: size, Preview: preview} + d = Doc{Owner: owner, KB: kb, Name: name, Ext: ext, Content: content, ObjectKey: objectKey, Size: size, Preview: preview} if err := p.db.WithContext(ctx).Create(&d).Error; err != nil { return "", "", err } @@ -137,7 +136,7 @@ func (p *Postgres) SaveDoc(ctx context.Context, owner, kb, name, ext, md5, conte return "", "", qerr } oldObjectKey = d.ObjectKey // 覆盖前的旧对象键 - d.Ext, d.MD5, d.Content, d.ObjectKey, d.Size, d.Preview = ext, md5, content, objectKey, size, preview + d.Ext, d.Content, d.ObjectKey, d.Size, d.Preview = ext, content, objectKey, size, preview if err := p.db.WithContext(ctx).Save(&d).Error; err != nil { return "", "", err } @@ -168,6 +167,20 @@ func (p *Postgres) GetDocByID(ctx context.Context, owner, id string) (*Doc, erro return &d, nil } +// DeleteDocByID 删除一份文件的 PG 痕迹（owner 作用域防越权）：doc 行 + 其出链/入链。 +// 入链（别处指向本文件）一并删，避免悬空脏链。文档不存在视作已删（返回 nil）。 +func (p *Postgres) DeleteDocByID(ctx context.Context, owner, kb, id string) error { + if p.db == nil { + return nil + } + return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Where("owner = ? AND kb = ? AND (from_id = ? OR to_id = ?)", owner, kb, id, id).Delete(&DocLink{}).Error; err != nil { + return err + } + return tx.Where("owner = ? AND id = ?", owner, id).Delete(&Doc{}).Error + }) +} + // ReplaceDocLinks 以源文件 ID 重建其出链（先删旧，再按 [[名称]] 解析目标 ID 后插新）—— 入库/编辑时调用。 // 目标文档尚未入库时 ToID 留空（悬空），待其入库时由 ResolveInboundLinks 回填。 func (p *Postgres) ReplaceDocLinks(ctx context.Context, owner, kb, fromID string, toNames []string) error {