Blizzard
a8c0bb42a7
把 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>
403 lines
15 KiB
Go
403 lines
15 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"strings"
|
||
"time"
|
||
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
)
|
||
|
||
// Space 是一个「工作区/协作空间」——共享的资源容器（增量3，见 SPACE_DESIGN.md）。
|
||
// 资源(Agent/KB…)挂在 Space 上而非 user 上，让"个人私有 / 项目临时组队 / 整租户共享"三种场景
|
||
// 出自同一模型。Space 属单一租户（隔离/计费边界仍在租户）。
|
||
// 与 Tenant/TenantMember 一样是"基础设施表"，不标 isTenantScoped（tenant 过滤在查询里显式做）。
|
||
type Space struct {
|
||
BaseModel
|
||
TenantID string `gorm:"size:64;index"`
|
||
Name string `gorm:"size:128"`
|
||
Kind string `gorm:"size:16;index"` // personal / project / tenant
|
||
Creator string `gorm:"size:64;index"` // 建者 user.id
|
||
Archived bool // 软归档（临时空间完事即归档，从切换器隐藏）
|
||
}
|
||
|
||
func (Space) TableName() string { return "sundynix_space" }
|
||
|
||
// SpaceMember 是空间成员关系（谁在这个空间 + 空间内角色）。复用租户角色语义（RoleRank 阶梯）。
|
||
type SpaceMember struct {
|
||
BaseModel
|
||
SpaceID string `gorm:"size:64;uniqueIndex:idx_space_user;index"`
|
||
UserID string `gorm:"size:64;uniqueIndex:idx_space_user;index"`
|
||
Role string `gorm:"size:16"`
|
||
Status string `gorm:"size:16"` // active / removed
|
||
}
|
||
|
||
func (SpaceMember) TableName() string { return "sundynix_space_member" }
|
||
|
||
// 空间类型常量。
|
||
const (
|
||
SpaceKindPersonal = "personal"
|
||
SpaceKindProject = "project"
|
||
SpaceKindTenant = "tenant"
|
||
)
|
||
|
||
// ---- 基础查询 ----
|
||
|
||
// tenantIDForSpace 返回某空间的租户 id（异步入库路径按 space 补 tenant_id 用）。
|
||
func (p *Postgres) tenantIDForSpace(ctx context.Context, spaceID string) string {
|
||
if p.db == nil || spaceID == "" {
|
||
return ""
|
||
}
|
||
var s Space
|
||
if err := p.db.WithContext(ctx).Select("tenant_id").First(&s, "id = ?", spaceID).Error; err != nil {
|
||
return ""
|
||
}
|
||
return s.TenantID
|
||
}
|
||
|
||
// GetSpace 按 id 取空间（不存在返回 nil）。
|
||
func (p *Postgres) GetSpace(ctx context.Context, id string) (*Space, error) {
|
||
if p.db == nil {
|
||
return nil, errStoreDisabled
|
||
}
|
||
var s Space
|
||
if err := p.db.WithContext(ctx).First(&s, "id = ?", id).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil, nil
|
||
}
|
||
return nil, err
|
||
}
|
||
return &s, nil
|
||
}
|
||
|
||
// SpaceMemberRole 返回用户在某空间的角色（无 active 成员关系返回空）。
|
||
func (p *Postgres) SpaceMemberRole(ctx context.Context, spaceID, userID string) string {
|
||
if p.db == nil {
|
||
return ""
|
||
}
|
||
var m SpaceMember
|
||
if err := p.db.WithContext(ctx).
|
||
Where("space_id = ? AND user_id = ? AND status = ?", spaceID, userID, "active").
|
||
First(&m).Error; err != nil {
|
||
return ""
|
||
}
|
||
return m.Role
|
||
}
|
||
|
||
// addSpaceMember 幂等加成员（命中唯一 (space_id,user_id) 则改角色并置 active，复活已移除者）。
|
||
func (p *Postgres) addSpaceMember(ctx context.Context, spaceID, userID, role string) error {
|
||
return p.db.WithContext(ctx).Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{{Name: "space_id"}, {Name: "user_id"}},
|
||
DoUpdates: clause.Assignments(map[string]any{"role": role, "status": "active", "updated_at": time.Now()}),
|
||
}).Create(&SpaceMember{SpaceID: spaceID, UserID: userID, Role: role, Status: "active"}).Error
|
||
}
|
||
|
||
// ---- 个人空间（迁移地基）----
|
||
|
||
// personalSpaceID 返回用户在某租户的个人空间 id（无则空）。
|
||
func (p *Postgres) personalSpaceID(ctx context.Context, tenantID, userID string) string {
|
||
var s Space
|
||
if err := p.db.WithContext(ctx).Select("id").
|
||
Where("tenant_id = ? AND creator = ? AND kind = ?", tenantID, userID, SpaceKindPersonal).
|
||
Order("created_at asc").First(&s).Error; err != nil {
|
||
return ""
|
||
}
|
||
return s.ID
|
||
}
|
||
|
||
// EnsurePersonalSpace 幂等保证用户在某租户有个人空间（无则建 + 建者为 owner 成员）。
|
||
func (p *Postgres) EnsurePersonalSpace(ctx context.Context, tenantID, userID string) (*Space, error) {
|
||
if p.db == nil {
|
||
return nil, errStoreDisabled
|
||
}
|
||
if id := p.personalSpaceID(ctx, tenantID, userID); id != "" {
|
||
return p.GetSpace(ctx, id)
|
||
}
|
||
s := &Space{TenantID: tenantID, Name: "个人空间", Kind: SpaceKindPersonal, Creator: userID}
|
||
if err := p.db.WithContext(ctx).Create(s).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
if err := p.addSpaceMember(ctx, s.ID, userID, RoleOwner); err != nil {
|
||
return nil, err
|
||
}
|
||
return s, nil
|
||
}
|
||
|
||
// ---- 活跃空间解析 / 切换 ----
|
||
|
||
// ActiveSpaceForUser 解析用户在某租户的活跃空间：user.active_space_id 若有效（存在、属该租户、且用户是
|
||
// active 成员）→ 用之；否则回退该租户的个人空间（并保证其存在）。SpaceContext 中间件每请求调用。
|
||
func (p *Postgres) ActiveSpaceForUser(ctx context.Context, tenantID, userID string) (*Space, error) {
|
||
def, err := p.EnsurePersonalSpace(ctx, tenantID, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var u User
|
||
if e := p.db.WithContext(ctx).Select("active_space_id").First(&u, "id = ?", userID).Error; e == nil {
|
||
if u.ActiveSpaceID != "" && u.ActiveSpaceID != def.ID {
|
||
if s, _ := p.GetSpace(ctx, u.ActiveSpaceID); s != nil && s.TenantID == tenantID && !s.Archived &&
|
||
p.SpaceMemberRole(ctx, s.ID, userID) != "" {
|
||
return s, nil
|
||
}
|
||
}
|
||
}
|
||
return def, nil
|
||
}
|
||
|
||
// SetActiveSpace 切换用户活跃空间（须为其 active 成员）。
|
||
func (p *Postgres) SetActiveSpace(ctx context.Context, userID, spaceID string) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
if p.SpaceMemberRole(ctx, spaceID, userID) == "" {
|
||
return errors.New("你不是该空间的成员")
|
||
}
|
||
return p.db.WithContext(ctx).Model(&User{}).Where("id = ?", userID).Update("active_space_id", spaceID).Error
|
||
}
|
||
|
||
// ---- 空间列表 / 创建 ----
|
||
|
||
// SpaceInfo 是空间列表一行（含当前用户角色 + 成员数）。
|
||
type SpaceInfo struct {
|
||
ID string `json:"id"`
|
||
TenantID string `json:"tenant_id"`
|
||
Name string `json:"name"`
|
||
Kind string `json:"kind"`
|
||
Creator string `json:"creator"`
|
||
Archived bool `json:"archived"`
|
||
Role string `json:"role"` // 当前用户在此空间的角色
|
||
Members int64 `json:"members"` // active 成员数
|
||
}
|
||
|
||
// MySpaces 返回某用户在某租户内所属（active 成员）的空间（供切换器；含角色 + 成员数）。
|
||
func (p *Postgres) MySpaces(ctx context.Context, tenantID, userID string) ([]SpaceInfo, error) {
|
||
if p.db == nil {
|
||
return nil, nil
|
||
}
|
||
var out []SpaceInfo
|
||
err := p.db.WithContext(ctx).Table("sundynix_space s").
|
||
Select("s.id, s.tenant_id, s.name, s.kind, s.creator, s.archived, m.role, "+
|
||
"(SELECT count(*) FROM sundynix_space_member m2 WHERE m2.space_id = s.id AND m2.status = 'active') as members").
|
||
Joins("JOIN sundynix_space_member m ON m.space_id = s.id").
|
||
Where("m.user_id = ? AND m.status = 'active' AND s.tenant_id = ? AND s.deleted_at IS NULL", userID, tenantID).
|
||
Order("s.kind = 'personal' desc, s.created_at asc").Scan(&out).Error
|
||
return out, err
|
||
}
|
||
|
||
// CreateSpace 在某租户建一个空间（建者自动成为 owner 成员）。kind 空则 project。
|
||
func (p *Postgres) CreateSpace(ctx context.Context, tenantID, name, kind, creator string) (*Space, error) {
|
||
if p.db == nil {
|
||
return nil, errStoreDisabled
|
||
}
|
||
if kind == "" {
|
||
kind = SpaceKindProject
|
||
}
|
||
if strings.TrimSpace(name) == "" {
|
||
return nil, errors.New("空间名必填")
|
||
}
|
||
s := &Space{TenantID: tenantID, Name: strings.TrimSpace(name), Kind: kind, Creator: creator}
|
||
if err := p.db.WithContext(ctx).Create(s).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
if err := p.addSpaceMember(ctx, s.ID, creator, RoleOwner); err != nil {
|
||
return nil, err
|
||
}
|
||
return s, nil
|
||
}
|
||
|
||
// ArchiveSpace 归档空间（软；个人空间不可归档）。
|
||
func (p *Postgres) ArchiveSpace(ctx context.Context, spaceID string) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
s, err := p.GetSpace(ctx, spaceID)
|
||
if err != nil || s == nil {
|
||
return errors.New("空间不存在")
|
||
}
|
||
if s.Kind == SpaceKindPersonal {
|
||
return errors.New("个人空间不可归档")
|
||
}
|
||
return p.db.WithContext(ctx).Model(&Space{}).Where("id = ?", spaceID).Update("archived", true).Error
|
||
}
|
||
|
||
// ---- 空间成员 CRUD（照 TenantMember 套路）----
|
||
|
||
// ListSpaceMembers 返回某空间成员（含邮箱/名字/角色），排除已移除。
|
||
func (p *Postgres) ListSpaceMembers(ctx context.Context, spaceID string) ([]MemberInfo, error) {
|
||
if p.db == nil {
|
||
return nil, nil
|
||
}
|
||
var out []MemberInfo
|
||
err := p.db.WithContext(ctx).Table("sundynix_space_member m").
|
||
Select("m.user_id, u.email, u.name, m.role, m.status, m.created_at as joined_at").
|
||
Joins("JOIN sundynix_user u ON u.id = m.user_id").
|
||
Where("m.space_id = ? AND m.status <> ?", spaceID, "removed").
|
||
Order("m.created_at asc").Scan(&out).Error
|
||
return out, err
|
||
}
|
||
|
||
// AddSpaceMemberByEmail 按邮箱把已注册用户加入空间（幂等；须对方先注册）。
|
||
// 调用方应先确保该用户是租户成员（空间成员应是租户成员的子集）。
|
||
func (p *Postgres) AddSpaceMemberByEmail(ctx context.Context, spaceID, email, role string) (*MemberInfo, error) {
|
||
if p.db == nil {
|
||
return nil, errStoreDisabled
|
||
}
|
||
if !ValidRole(role) {
|
||
return nil, errors.New("非法角色")
|
||
}
|
||
u, err := p.GetUserByEmail(ctx, strings.ToLower(strings.TrimSpace(email)))
|
||
if errors.Is(err, gorm.ErrRecordNotFound) || u == nil {
|
||
return nil, errors.New("该邮箱尚未注册，请对方先注册后再加入")
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s, err := p.GetSpace(ctx, spaceID)
|
||
if err != nil || s == nil {
|
||
return nil, errors.New("空间不存在")
|
||
}
|
||
if p.MemberRole(ctx, s.TenantID, u.ID) == "" {
|
||
return nil, errors.New("对方不是本租户成员，请先加入租户")
|
||
}
|
||
if err := p.addSpaceMember(ctx, spaceID, u.ID, role); err != nil {
|
||
return nil, err
|
||
}
|
||
return &MemberInfo{UserID: u.ID, Email: u.Email, Name: u.Name, Role: role, Status: "active"}, nil
|
||
}
|
||
|
||
// SetSpaceMemberRole 改空间成员角色（不能改/授 owner）。
|
||
func (p *Postgres) SetSpaceMemberRole(ctx context.Context, spaceID, userID, role string) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
if !ValidRole(role) {
|
||
return errors.New("非法角色")
|
||
}
|
||
if p.SpaceMemberRole(ctx, spaceID, userID) == RoleOwner {
|
||
return errors.New("不能改 owner 的角色")
|
||
}
|
||
if role == RoleOwner {
|
||
return errors.New("不能直接授予 owner")
|
||
}
|
||
return p.db.WithContext(ctx).Model(&SpaceMember{}).
|
||
Where("space_id = ? AND user_id = ?", spaceID, userID).Update("role", role).Error
|
||
}
|
||
|
||
// RemoveSpaceMember 软移除空间成员（不能移除 owner）。
|
||
func (p *Postgres) RemoveSpaceMember(ctx context.Context, spaceID, userID string) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
if p.SpaceMemberRole(ctx, spaceID, userID) == RoleOwner {
|
||
return errors.New("不能移除 owner")
|
||
}
|
||
return p.db.WithContext(ctx).Model(&SpaceMember{}).
|
||
Where("space_id = ? AND user_id = ?", spaceID, userID).Update("status", "removed").Error
|
||
}
|
||
|
||
// ---- 迁移回填（启动时幂等调用）----
|
||
|
||
// BackfillPersonalSpaces 给每个 active 租户成员补建个人空间（幂等）。返回补建数。
|
||
func (p *Postgres) BackfillPersonalSpaces(ctx context.Context) (int, error) {
|
||
if p.db == nil {
|
||
return 0, errStoreDisabled
|
||
}
|
||
var members []TenantMember
|
||
if err := p.db.WithContext(ctx).Where("status = ?", "active").Find(&members).Error; err != nil {
|
||
return 0, err
|
||
}
|
||
n := 0
|
||
for _, m := range members {
|
||
if p.personalSpaceID(ctx, m.TenantID, m.UserID) != "" {
|
||
continue
|
||
}
|
||
if _, err := p.EnsurePersonalSpace(ctx, m.TenantID, m.UserID); err != nil {
|
||
return n, err
|
||
}
|
||
n++
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// MigrateAgentSpaces 把存量 Agent 的 space_id 回填为其 (tenant_id, owner) 个人空间，
|
||
// 然后 DROP 旧唯一索引 idx_agent_on(owner,name) 并建新唯一索引 idx_agent_sn(space_id,name)。
|
||
// 顺序关键：先回填再建唯一索引，否则存量 space_id 空会撞车。全程幂等。
|
||
func (p *Postgres) MigrateAgentSpaces(ctx context.Context) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
// 回填 space_id：按 agent.tenant_id + agent.owner 命中个人空间。
|
||
sub := "SELECT id FROM sundynix_space s WHERE s.tenant_id = a.tenant_id AND s.creator = a.owner AND s.kind = 'personal' ORDER BY s.created_at ASC LIMIT 1"
|
||
if err := p.db.WithContext(ctx).Exec(
|
||
"UPDATE sundynix_agent a SET space_id = ("+sub+") WHERE (a.space_id IS NULL OR a.space_id = '') AND a.owner <> ''",
|
||
).Error; err != nil {
|
||
return err
|
||
}
|
||
// 旧唯一索引 DROP（AutoMigrate 不删；留着会过约束"同人跨空间不能重名"）。
|
||
if err := p.db.WithContext(ctx).Exec("DROP INDEX IF EXISTS idx_agent_on").Error; err != nil {
|
||
return err
|
||
}
|
||
// 新唯一索引（回填后建，安全）。
|
||
if err := p.db.WithContext(ctx).Exec(
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_sn ON sundynix_agent(space_id, name)",
|
||
).Error; err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// DocForRestorage 是存储层重灌迁移用的一条文档视图（拿回原文所需的键）。
|
||
type DocForRestorage struct {
|
||
ID string
|
||
SpaceID string
|
||
Owner string
|
||
KB string
|
||
Name string
|
||
Content string
|
||
ObjectKey string
|
||
}
|
||
|
||
// DocsForRestorage 列出全部有 space_id 的存量文档（系统级跨租户；供存储层 owner/kb→space/kb 重灌）。
|
||
func (p *Postgres) DocsForRestorage(ctx context.Context) ([]DocForRestorage, error) {
|
||
if p.db == nil {
|
||
return nil, errStoreDisabled
|
||
}
|
||
var rows []DocForRestorage
|
||
err := p.db.WithContext(WithoutTenant(ctx)).Table("sundynix_doc").
|
||
Select("id, space_id, owner, kb, name, content, object_key").
|
||
Where("space_id <> '' AND owner <> ''").Order("created_at asc").Scan(&rows).Error
|
||
return rows, err
|
||
}
|
||
|
||
// MigrateKBSpaces 把存量 KB/Doc/DocLink 的 space_id 回填为其 (tenant_id, owner) 个人空间，
|
||
// 再 DROP 旧唯一索引、建新唯一索引（同 MigrateAgentSpaces 的顺序坑规避）。全程幂等。
|
||
// 注：仅迁移 PG 元数据作用域；Milvus/Bleve/Neo4j 的存量向量/全文/图谱另由 kb 存储重灌迁移。
|
||
func (p *Postgres) MigrateKBSpaces(ctx context.Context) error {
|
||
if p.db == nil {
|
||
return errStoreDisabled
|
||
}
|
||
sub := "SELECT id FROM sundynix_space s WHERE s.tenant_id = t.tenant_id AND s.creator = t.owner AND s.kind = 'personal' ORDER BY s.created_at ASC LIMIT 1"
|
||
for _, tbl := range []string{"sundynix_kb", "sundynix_doc", "sundynix_doc_link"} {
|
||
if err := p.db.WithContext(ctx).Exec(
|
||
"UPDATE "+tbl+" t SET space_id = ("+sub+") WHERE (t.space_id IS NULL OR t.space_id = '') AND t.owner <> ''",
|
||
).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
// 旧唯一索引 DROP + 新唯一索引建（回填后，安全）。DocLink 无唯一约束，不涉及。
|
||
stmts := []string{
|
||
"DROP INDEX IF EXISTS idx_kb_owner_name",
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_kb_sn ON sundynix_kb(space_id, name)",
|
||
"DROP INDEX IF EXISTS idx_doc_okn",
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_doc_skn ON sundynix_doc(space_id, kb, name)",
|
||
}
|
||
for _, s := range stmts {
|
||
if err := p.db.WithContext(ctx).Exec(s).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|