Files
sundynix-agentix/sundynix-gateway/internal/store/model.go
T
Blizzard a8c0bb42a7 feat(space): 共享工作区增量3b —— KB 知识库按 Space 共享(存储 re-key)
把 3a 的 Space 作用域推到 KB 层:KB/Doc/DocLink owner→space_id,作用域键
owner/name→space_id/name,同空间成员共享知识库(检索/入库/文库/双链/图谱)。

后端:
- store: KB/Doc/DocLink 加 space_id,唯一索引 (owner,*)→(space_id,*),owner 降级创建人;
  查询全改 space 作用域;SaveDoc/ListVault/GetDocByID/DeleteDocByID/ReplaceDocLinks/
  ResolveInboundLinks/ListLinks 改 space;tenantIDForSpace 补异步入库租户
- MigrateKBSpaces 启动迁移(space_id 回填 + 唯一索引换新,同 Agent 顺序坑规避)
- scopedKB owner/name→space_id/name;IngestJob 契约加 SpaceID;enqueueIngest/runIngest
  穿 space;MinIO 对象键改 space/kb/doc(避免跨空间同名撞键,老键不透明不迁)
- KB 写路由(create/ingest/ingest_file/note/delete)挂 RequireSpaceRole(member):viewer 只读
- 存储层重灌迁移端点 POST /admin/migrate-kb-storage(异步):为存量文档入队新 space
  作用域的重灌作业(复用 JetStream 入库 worker 池),先删旧键;避免同步重嵌撑爆 HTTP 超时

桌面端:
- KbView 收 spaceId(变则重拉库)+spaceReadOnly(viewer 禁建库/入库/文件/笔记);VaultPanel 同

验证(gateway+mcp-go+Milvus/Neo4j/embedding 全栈):
- PG 迁移: 20/21 KB + 50/54 doc 回填 space_id(4 未迁=pre-多租户 owner='wt' 空租户遗留,
  正确跳过),唯一索引 idx_kb_sn/idx_doc_skn 换新、旧索引删除
- KB 共享: member 见共享库 / viewer 建库·入库 403 / 切回个人空间隔离(看不到)
- 全向量链路: RagA 入库(真 dashscope embedding)→ RagB(空间member)检索命中 RagA 内容
- 存储重灌: 端点异步入队 49 作业(worker 池背压处理),重灌后老文档在新 space 键可检索

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 13:00:34 +08:00

419 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package store
import (
"context"
"errors"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"github.com/sundynix/sundynix-shared/contract"
)
// KB 是一个知识库(增量3:按 space 共享 + 按 kind 组织:文件夹/项目/案件/通用)。
// 表名 sundynix_kb。作用域从 owner 改为 space_id——(space_id,name) 唯一,同一工作区内库名不重复;
// owner 降级为创建人。向量/全文/图谱以 "space_id/name" 作分区键(见 handler.scopedKB)。
// 唯一索引不在结构体标记里建(AutoMigrate 早于回填会撞车),由 MigrateKBSpaces 回填后显式建。
type KB struct {
BaseModel
TenantID string `gorm:"size:64;index"` // 多租户作用域(gorm tenant 插件按上下文自动填/过滤)
SpaceID string `gorm:"size:64;index"` // 所属工作区(共享单位;唯一约束见 MigrateKBSpaces
Owner string `gorm:"size:64;index"` // 创建人 user.id(归属,非隔离键)
Name string `gorm:"size:64"`
Kind string `gorm:"size:16"` // folder / project / case / general
}
func (KB) TableName() string { return "sundynix_kb" }
func (KB) isTenantScoped() {}
// ListKB 列出某工作区的全部知识库(按创建时间;tenant 插件仍按 ctx 租户过滤)。
func (p *Postgres) ListKB(ctx context.Context, spaceID string) ([]KB, error) {
if p.db == nil {
return nil, nil
}
var rows []KB
err := p.db.WithContext(ctx).Where("space_id = ?", spaceID).Order("id").Find(&rows).Error
return rows, err
}
// EnsureKB 幂等登记一个知识库到某工作区(已存在则保持,不覆盖 kind;owner 记创建人)。
func (p *Postgres) EnsureKB(ctx context.Context, spaceID, owner, name, kind string) error {
if p.db == nil {
return nil // 降级模式:不持久化注册表,不阻断入库
}
if kind == "" {
kind = "general"
}
return p.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "space_id"}, {Name: "name"}},
DoNothing: true,
}).Create(&KB{SpaceID: spaceID, Owner: owner, Name: name, Kind: kind}).Error
}
// Agent 是一份保存的 Agent 编排(React Flow 图 JSON)。
// 表名 sundynix_agent。增量3:作用域从 owner 改为 space_id——(space_id,name) 唯一,
// 同一工作区内编排名不重复;owner 降级为"创建人"归属(谁建的,供 UI 显示 / 删他人鉴权)。
// 唯一索引 idx_agent_sn(space_id,name) 不在结构体标记里建(否则 AutoMigrate 早于回填、存量 space_id 空会撞车),
// 而由启动迁移 MigrateAgentSpaces 在回填 space_id 后显式建;旧 idx_agent_on(owner,name) 同步 DROP。
type Agent struct {
BaseModel
TenantID string `gorm:"size:64;index"` // 多租户作用域(tenant 插件自动填/过滤,防越权写他租)
SpaceID string `gorm:"size:64;index"` // 所属工作区(共享单位;唯一约束见 MigrateAgentSpaces
Owner string `gorm:"size:64;index"` // 创建人 user.id(归属,非隔离键)
Name string `gorm:"size:128"`
Graph string `gorm:"type:text"` // {nodes,edges} 的 JSON(含布局)
}
func (Agent) TableName() string { return "sundynix_agent" }
func (Agent) isTenantScoped() {}
// AgentInfo 是编排列表视图(含创建人邮箱/名字,供共享工作区显示"谁建的")。
type AgentInfo struct {
Name string `json:"name"`
Graph string `json:"graph"`
Owner string `json:"owner"` // 创建人 user.id
CreatorName string `json:"creator_name"`
CreatorMail string `json:"creator_mail"`
UpdatedAt time.Time `json:"updated_at"`
}
// ListAgents 返回某工作区(space)的全部编排(最近更新在前,join 创建人名字)。
// tenant 插件仍会按 ctx 租户过滤(防越权);空间是主作用域。
func (p *Postgres) ListAgents(ctx context.Context, spaceID string) ([]AgentInfo, error) {
if p.db == nil {
return nil, nil
}
var out []AgentInfo
err := p.db.WithContext(ctx).Table("sundynix_agent a").
Select("a.name, a.graph, a.owner, u.name as creator_name, u.email as creator_mail, a.updated_at").
Joins("LEFT JOIN sundynix_user u ON u.id = a.owner").
Where("a.space_id = ? AND a.deleted_at IS NULL", spaceID).
Order("a.updated_at desc").Scan(&out).Error
return out, err
}
// SaveAgent 新建/更新一份编排(space_id+name 唯一,重名覆盖图与更新时间;owner 首建时记创建人,覆盖不改)。
func (p *Postgres) SaveAgent(ctx context.Context, spaceID, creator, name, graph string) error {
if p.db == nil {
return errStoreDisabled
}
return p.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "space_id"}, {Name: "name"}},
DoUpdates: clause.Assignments(map[string]any{"graph": graph, "updated_at": time.Now()}),
}).Create(&Agent{SpaceID: spaceID, Owner: creator, Name: name, Graph: graph}).Error
}
// GetAgent 取某工作区内一份编排(含 owner,供删他人鉴权判定)。
func (p *Postgres) GetAgent(ctx context.Context, spaceID, name string) (*Agent, error) {
if p.db == nil {
return nil, nil
}
var a Agent
if err := p.db.WithContext(ctx).Where("space_id = ? AND name = ?", spaceID, name).First(&a).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, err
}
return &a, nil
}
// DeleteAgent 删除某工作区的一份编排。
func (p *Postgres) DeleteAgent(ctx context.Context, spaceID, name string) error {
if p.db == nil {
return errStoreDisabled
}
return p.db.WithContext(ctx).Where("space_id = ? AND name = ?", spaceID, name).Delete(&Agent{}).Error
}
// Doc 是入库的一份文件/笔记 —— 文件主表(供 Obsidian 式"文库"浏览:列表 + Markdown 阅读 + 双链)。
// 表名 sundynix_doc。增量3:作用域 owner→space_id(space_id,kb,name) 唯一;owner 降级为创建人(上传者)。
// 唯一索引由 MigrateKBSpaces 回填后显式建(AutoMigrate 早于回填会撞车)。
type Doc struct {
BaseModel
TenantID string `gorm:"size:64;index"` // 多租户作用域(入库异步无请求 ctx,SaveDoc 按 owner 补)
SpaceID string `gorm:"size:64;index"` // 所属工作区(共享单位;唯一约束见 MigrateKBSpaces
Owner string `gorm:"size:64;index"` // 创建人/上传者 user.id(归属,非隔离键)
KB string `gorm:"size:64;index"`
Name string `gorm:"size:160"` // 文件名称(不含扩展名 / 笔记名)
Ext string `gorm:"size:16"` // 文件后缀(.md/.pdf/.docx…;笔记/文本为空)
Size int // 原文字数(rune
Preview string `gorm:"size:600"` // 前若干字预览(列表/反链用,不拉全文)
Content string `gorm:"type:text"` // 正文一律落 MinIO 后置空;仅 MinIO 不可用时回退内联存这里
ObjectKey string `gorm:"size:160"` // 正文在 MinIO 的对象键(常态非空;空=回退内联 Content)
}
func (Doc) TableName() string { return "sundynix_doc" }
func (Doc) isTenantScoped() {}
// DocLink 是文档间 [[双链]] 的索引(space+kb 内 from→to),以 Doc.ID 关联,供反链/笔记关系图按 SQL 查询。
// 增量3:作用域 owner→space_id。入库/编辑时按 from 文档重建其出链;目标尚未入库时 ToID 为空(悬空)。
type DocLink struct {
BaseModel
TenantID string `gorm:"size:64;index"` // 多租户作用域(ReplaceDocLinks 按 space 的租户补)
SpaceID string `gorm:"size:64;index:idx_link_sf"`
Owner string `gorm:"size:64;index"` // 创建人(归属)
KB string `gorm:"size:64;index:idx_link_sf"`
FromID string `gorm:"size:24;index:idx_link_sf"` // 源文档 Doc.ID
ToID string `gorm:"size:24;index"` // 目标文档 Doc.ID(空=悬空:目标尚未入库)
ToName string `gorm:"size:160"` // [[原始名]],供悬空链接展示 / 目标入库后回填 ToID
}
func (DocLink) TableName() string { return "sundynix_doc_link" }
func (DocLink) isTenantScoped() {}
// SaveDoc 写入/更新一份文件(space_id+kb+name 唯一,重名覆盖),返回 (文件雪花 ID, 被覆盖的旧对象键)。
// space_id 是作用域,owner 记创建人(上传者)。tenant_id 从 space 的租户补(异步入库无请求 ctx 租户)。
func (p *Postgres) SaveDoc(ctx context.Context, spaceID, owner, kb, name, ext, content, objectKey string, size int, preview string) (id, oldObjectKey string, err error) {
if p.db == nil {
return "", "", nil
}
tenantID := p.tenantIDForSpace(ctx, spaceID)
var d Doc
qerr := p.db.WithContext(ctx).Where("space_id = ? AND kb = ? AND name = ?", spaceID, kb, name).First(&d).Error
if errors.Is(qerr, gorm.ErrRecordNotFound) {
d = Doc{TenantID: tenantID, SpaceID: spaceID, 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
}
return d.ID, "", nil
}
if qerr != nil {
return "", "", qerr
}
oldObjectKey = d.ObjectKey // 覆盖前的旧对象键
if tenantID != "" {
d.TenantID = tenantID // 覆盖路径也补齐(老行可能为空);查不到则不动,别把已有值抹空
}
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
}
return d.ID, oldObjectKey, nil
}
// ListVault 返回文库列表(仅元数据 + 预览,不含全文),避免一次拉回整库正文。
func (p *Postgres) ListVault(ctx context.Context, spaceID, kb string) ([]Doc, error) {
if p.db == nil {
return nil, nil
}
var rows []Doc
err := p.db.WithContext(ctx).
Select("id", "name", "ext", "size", "preview", "object_key", "owner", "updated_at").
Where("space_id = ? AND kb = ?", spaceID, kb).Order("updated_at desc").Find(&rows).Error
return rows, err
}
// GetDocByID 按文件 ID 取单篇文档(含全文 Content 与 ObjectKey),space 作用域防越权。
func (p *Postgres) GetDocByID(ctx context.Context, spaceID, id string) (*Doc, error) {
if p.db == nil {
return nil, nil
}
var d Doc
if err := p.db.WithContext(ctx).Where("space_id = ? AND id = ?", spaceID, id).First(&d).Error; err != nil {
return nil, err
}
return &d, nil
}
// DeleteDocByID 删除一份文件的 PG 痕迹(space 作用域防越权):doc 行 + 其出链/入链。
func (p *Postgres) DeleteDocByID(ctx context.Context, spaceID, 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("space_id = ? AND kb = ? AND (from_id = ? OR to_id = ?)", spaceID, kb, id, id).Delete(&DocLink{}).Error; err != nil {
return err
}
return tx.Where("space_id = ? AND id = ?", spaceID, id).Delete(&Doc{}).Error
})
}
// ReplaceDocLinks 以源文件 ID 重建其出链(先删旧,再按 [[名称]] 解析目标 ID 后插新)—— 入库/编辑时调用。
func (p *Postgres) ReplaceDocLinks(ctx context.Context, spaceID, owner, kb, fromID string, toNames []string) error {
if p.db == nil {
return nil
}
tenantID := p.tenantIDForSpace(ctx, spaceID)
return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("space_id = ? AND kb = ? AND from_id = ?", spaceID, kb, fromID).Delete(&DocLink{}).Error; err != nil {
return err
}
for _, name := range toNames {
if name == "" {
continue
}
var t Doc
toID := ""
if e := tx.Select("id").Where("space_id = ? AND kb = ? AND name = ?", spaceID, kb, name).First(&t).Error; e == nil {
toID = t.ID
}
if toID == fromID { // 自链跳过
continue
}
if err := tx.Create(&DocLink{TenantID: tenantID, SpaceID: spaceID, Owner: owner, KB: kb, FromID: fromID, ToID: toID, ToName: name}).Error; err != nil {
return err
}
}
return nil
})
}
// ResolveInboundLinks 把指向 name 的悬空链接(ToID 空)回填为 id —— 目标文档入库后调用。
func (p *Postgres) ResolveInboundLinks(ctx context.Context, spaceID, kb, name, id string) error {
if p.db == nil {
return nil
}
return p.db.WithContext(ctx).Model(&DocLink{}).
Where("space_id = ? AND kb = ? AND to_name = ? AND (to_id = '' OR to_id IS NULL)", spaceID, kb, name).
Update("to_id", id).Error
}
// ListLinks 返回某 kb 已解析(两端均为本库文件)的 [[双链]](FromID→ToID),供反链/笔记关系图按 ID 渲染。
func (p *Postgres) ListLinks(ctx context.Context, spaceID, kb string) ([]DocLink, error) {
if p.db == nil {
return nil, nil
}
var rows []DocLink
err := p.db.WithContext(ctx).Where("space_id = ? AND kb = ? AND to_id <> ''", spaceID, kb).Find(&rows).Error
return rows, err
}
// Pricing 是某模型的计价配置(token↔真钱):按模型分输入/输出单价(每 1K token)。
// 表名 sundynix_pricing。ModelID 关联 sundynix_model.id,唯一。
type Pricing struct {
BaseModel
ModelID string `gorm:"size:24;uniqueIndex"` // 关联 sundynix_model.id
InputPer1K float64 `gorm:"column:input_per_1k"` // 每 1K 输入 token 单价
OutputPer1K float64 `gorm:"column:output_per_1k"` // 每 1K 输出 token 单价
Currency string `gorm:"size:8"` // 币种(CNY / USD…)
CreditWeight float64 `gorm:"column:credit_weight"` // 积分权重(每 token 烧积分的倍率;0/缺省按 1.0 计)
}
func (Pricing) TableName() string { return "sundynix_pricing" }
// ListPricing 列出全部计价配置。
func (p *Postgres) ListPricing(ctx context.Context) ([]Pricing, error) {
if p.db == nil {
return nil, nil
}
var rows []Pricing
err := p.db.WithContext(ctx).Find(&rows).Error
return rows, err
}
// UpsertPricing 写入/更新某模型的计价(model_id 唯一,重复即覆盖单价/币种/积分权重)。
// creditWeight 为每模型积分权重(0=按 1.0 计,即不加权)。
func (p *Postgres) UpsertPricing(ctx context.Context, modelID string, inPer1K, outPer1K, creditWeight float64, currency string) error {
if p.db == nil {
return errStoreDisabled
}
return p.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "model_id"}},
DoUpdates: clause.Assignments(map[string]any{"input_per_1k": inPer1K, "output_per_1k": outPer1K, "credit_weight": creditWeight, "currency": currency, "updated_at": time.Now()}),
}).Create(&Pricing{ModelID: modelID, InputPer1K: inPer1K, OutputPer1K: outPer1K, CreditWeight: creditWeight, Currency: currency}).Error
}
// LLMModel 是一个模型后端配置(控制面:管理员在此登记可用模型)。
// 表名 sundynix_model(遵守前缀约定)。每个 kind 同一时刻仅一条 Active=true。
type LLMModel struct {
BaseModel
Kind string `gorm:"size:16;index"` // chat / embedding
Provider string `gorm:"size:32"` // openai-compatible / vllm
BaseURL string `gorm:"size:255"` // 如 https://api.deepseek.com
APIKey string `gorm:"size:255"`
Model string `gorm:"size:64"` // 如 deepseek-chat / text-embedding-v3
Active bool
}
func (LLMModel) TableName() string { return "sundynix_model" }
// ListModels 列出某 kind 的模型配置(kind 空则全部)。
func (p *Postgres) ListModels(ctx context.Context, kind string) ([]LLMModel, error) {
if p.db == nil {
return nil, nil
}
var rows []LLMModel
q := p.db.WithContext(ctx).Order("id")
if kind != "" {
q = q.Where("kind = ?", kind)
}
err := q.Find(&rows).Error
return rows, err
}
// SaveModel 新增或更新一条模型配置(ID 空则新增,BeforeCreate 生成雪花 ID)。
func (p *Postgres) SaveModel(ctx context.Context, m *LLMModel) error {
if p.db == nil {
return errStoreDisabled
}
if m.ID == "" {
return p.db.WithContext(ctx).Create(m).Error
}
// 更新仅覆盖可编辑列:Active 由 SetActiveModel 单独管理,CreatedAt 不动——
// 避免整行 Save 把 active 清零(改 key/base_url 时不应顺手取消激活)。
return p.db.WithContext(ctx).Model(&LLMModel{}).Where("id = ?", m.ID).
Select("Kind", "Provider", "BaseURL", "APIKey", "Model").Updates(m).Error
}
// SetActiveModel 把指定模型设为激活(同 kind 内其余取消),事务保证每 kind 唯一激活。
func (p *Postgres) SetActiveModel(ctx context.Context, id string) error {
if p.db == nil {
return errStoreDisabled
}
return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var m LLMModel
if err := tx.First(&m, "id = ?", id).Error; err != nil {
return err
}
if err := tx.Model(&LLMModel{}).Where("kind = ? AND active = ?", m.Kind, true).Update("active", false).Error; err != nil {
return err
}
return tx.Model(&LLMModel{}).Where("id = ?", id).Update("active", true).Error
})
}
// GetActiveModel 返回某 kind 当前激活模型(无则 nil)。
func (p *Postgres) GetActiveModel(ctx context.Context, kind string) (*LLMModel, error) {
if p.db == nil {
return nil, nil
}
var m LLMModel
err := p.db.WithContext(ctx).Where("kind = ? AND active = ?", kind, true).First(&m).Error
if err != nil {
return nil, nil // 未配置激活模型
}
return &m, nil
}
// ActiveConfig 取某 kind 的激活模型配置;chat 还把"其它已登记 chat 模型"按序填进 Fallbacks
// 供 dispatcher 串成 failover 链(主 provider 抖动/挂掉时自动切备,平台不整体宕)。
func (p *Postgres) ActiveConfig(ctx context.Context, kind string) *contract.ModelConfig {
row, _ := p.GetActiveModel(ctx, kind)
if row == nil {
return nil
}
cfg := &contract.ModelConfig{Provider: row.Provider, BaseURL: row.BaseURL, APIKey: row.APIKey, Model: row.Model}
if kind == contract.ConfigKindChat {
all, _ := p.ListModels(ctx, kind)
for _, m := range all {
if m.ID == row.ID {
continue // 跳过激活模型(它已是主)
}
cfg.Fallbacks = append(cfg.Fallbacks, contract.ModelConfig{
Provider: m.Provider, BaseURL: m.BaseURL, APIKey: m.APIKey, Model: m.Model,
})
}
}
return cfg
}
// DeleteModel 删除一条模型配置。
func (p *Postgres) DeleteModel(ctx context.Context, id string) error {
if p.db == nil {
return errStoreDisabled
}
return p.db.WithContext(ctx).Delete(&LLMModel{}, "id = ?", id).Error
}