feat(release): wiki 通道按归属判定替换，共享知识库不再被整库清空
WeKnora 的 wiki 图谱与页面同表（wiki_pages），旧的库级 DELETE + 灌包在共享真实 知识库上会把其他文档的页面与 agent 手写页一起抹掉（2026-10-08 生产演练现场）。 - 新增 internal/releaseadapter/v071/wiki_scope.go：按 source_refs 判定归属 （全部引用属于本包 = InScope；引用包外知识 = 外来页；无引用 = index/log/ agent 页永不触碰），目标库存在外来页时降级为只替换包内页集，外来页上仅摘掉 本包知识引用，包内库级页不写；包内会写入的页与外来页同 slug 时 fail-closed 拒绝发布且不落任何改动。 - Replace 与 ReplaceWiki（撤权/回滚）共用同一套判定，不再整库清空。 - 作用域以一行日志落盘：release wiki replace scope in_scope=… degraded=… - 测试：单测 7 条（纯函数判定）+ 真库集成 3 条（共享库保护 / slug 冲突拒绝 / 镜像库整库替换不变），集成用例只用带随机后缀的临时知识库并自清理。 - 文档：P6 新增「wiki 通道的归属门禁」，P8 补验收点与镜像前提。
This commit is contained in:
姚章浩
@@ -256,6 +256,26 @@ tenant_id + knowledge_base_id + knowledge_id
|
||||
- Chunk seq_id 冲突时整个事务失败，不静默改 ID。
|
||||
- 不调用 WeKnora 解析和 Embedding。
|
||||
|
||||
### wiki 通道的归属门禁（2026-10-08 起）
|
||||
|
||||
WeKnora 的 wiki 图谱与页面同表（`wiki_pages`，节点/边靠 slug + `in_links`/`out_links` 表达），
|
||||
所以"库级快照整库替换"在**共享真实知识库**上会把其他文档的页面和 agent 手写页一起抹掉。
|
||||
导入端现在先判定归属，再决定写入范围：
|
||||
|
||||
- 归属来自页面 `source_refs`（形如 `<knowledge_id>|<doc_title>`；`chunk_refs` 只有 chunk UUID，
|
||||
不带知识 id）。一页引用的知识**全部属于本包**才算 `InScope`；引用包外知识的是外来页；完全没有
|
||||
引用的页（`index`/`log`、agent 手写页）永不触碰。
|
||||
- **目标库没有外来页** → 镜像库对原件的前提成立，保持整库替换语义（行为与旧实现一致，仍不删
|
||||
"无引用且不在包内"的页）。
|
||||
- **目标库存在外来页** → 自动降级为"只替换包内页集"：按 slug 删除重建包内页、只删包内 folder id、
|
||||
外来页上仅摘掉本包知识的引用（正文不动）、包内库级页（`index`/`log`）不写以免毁掉目标库自己的聚合。
|
||||
作用域以一行日志落盘：`release wiki replace scope in_scope=… foreign=… foreign_ref_stripped=…
|
||||
untouched=… skipped_library=… degraded=…`。
|
||||
- **包内会写入的页与外来页同 slug** → 拒绝发布（fail-closed），事务不落任何改动，错误信息点名该 slug。
|
||||
- 撤权（`delete`）与回滚走同一套判定（`ReplaceWiki`），不再整库清空。
|
||||
- 已知取舍：降级路径上共享页的正文仍由目标库的库级合成结果决定，无法在不重算（不额外烧 token）的
|
||||
前提下与本包一致；这类页只做引用级差集，并在作用域日志里计入 `foreign_ref_stripped`。
|
||||
|
||||
## Elasticsearch 通道
|
||||
|
||||
源导出：
|
||||
|
||||
@@ -59,4 +59,11 @@ target.KnowledgeBaseID == mapping.TargetKnowledgeBaseID （目标
|
||||
1. `go test ./...`、`go vet`、前端 `tsc` + `vitest` 全绿；迁移 44 up/down 往返可用。
|
||||
2. 单元/集成：映射生效后 `knowledges/chunks/wiki_pages` 的 KB 列、ES 文档字段均为目标库；摘要随改写重算；未绑定退回同 ID。
|
||||
3. 界面：系统设置可选目标库（列真实库名），保存时对目标 WeKnora 真校验（无效 ID 拒绝）；发布管理显示本次目标库。
|
||||
4. 真实跨环境：1 份文档发布到目标库 → 三通道读回一致 → 官方 API 检索命中 → 撤权后零命中 → 回滚恢复。
|
||||
4. 真实跨环境：1 份文档发布到目标库 → 三通道读回一致 → 官方 API 检索命中 → 撤权后零命中 → 回滚恢复。
|
||||
5. **wiki 归属门禁（2026-10-08 起）**：目标库存在引用包外知识的页时，发布自动降级为"只替换包内页集"，
|
||||
外来页 / agent 手写页 / 目标库自己的 `index`-`log` 一行不动；包内会写入的页与外来页同 slug 时拒绝发布
|
||||
且事务不留任何改动。真库集成用例：`TestWikiReplaceOnSharedKnowledgeBaseKeepsForeignPages`、
|
||||
`TestWikiReplaceRefusesWhenPackageSlugBelongsToForeignPage`、`TestWikiReplaceOnFaithfulMirrorReplacesWholeLibrary`
|
||||
（`TEST_WEKNORA_V071_TARGET_DATABASE_URL` 指向 v0.7.1 库时运行，只使用带随机后缀的临时知识库并自清理）。
|
||||
6. 跨库对发布的前置条件：**测试库与生产库承载同一批文档**（镜像）时 wiki 才能保持"测试=生产"；
|
||||
只做单文档发布时，共享页正文不保证两边一致（见 P6「wiki 通道的归属门禁」）。
|
||||
@@ -119,29 +119,25 @@ WHERE (id = $1 OR handle = $2)
|
||||
{`DELETE FROM knowledge_tag_relations WHERE knowledge_id = $1`, []any{identity.KnowledgeID}},
|
||||
{`DELETE FROM chunks WHERE tenant_id = $1 AND knowledge_base_id = $2 AND knowledge_id = $3`, []any{identity.TenantID, identity.KnowledgeBaseID, identity.KnowledgeID}},
|
||||
}
|
||||
if hasWiki {
|
||||
// wiki 是库级快照：整库替换（幂等），不做单文档 delta。
|
||||
clearStatements = append(clearStatements,
|
||||
clearStatement{`DELETE FROM wiki_pages WHERE tenant_id = $1 AND knowledge_base_id = $2`, []any{identity.TenantID, identity.KnowledgeBaseID}},
|
||||
clearStatement{`DELETE FROM wiki_folders WHERE tenant_id = $1 AND knowledge_base_id = $2`, []any{identity.TenantID, identity.KnowledgeBaseID}},
|
||||
)
|
||||
}
|
||||
for _, statement := range clearStatements {
|
||||
if _, err := tx.Exec(ctx, statement.sql, statement.args...); err != nil {
|
||||
return errors.New("clear prior target document rows")
|
||||
}
|
||||
}
|
||||
insertOrder := []string{"knowledges", "resources", "chunks", "knowledge_tags", "knowledge_tag_relations", "resource_bindings"}
|
||||
if hasWiki {
|
||||
// 文件夹先于页面：页面通过 folder_id 引用文件夹。
|
||||
insertOrder = append(insertOrder, "wiki_folders", "wiki_pages")
|
||||
}
|
||||
for _, tableName := range insertOrder {
|
||||
table := prepared.Table(tableName)
|
||||
if err := insertSnapshotRows(ctx, tx, table); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// wiki 是库级快照，但不能盲删整库：先按 source_refs 判定归属，只替换本包
|
||||
// 拥有的页面，其余（其他文档、agent 手写页）一行不动。
|
||||
if hasWiki {
|
||||
if _, err := importer.replaceWiki(ctx, tx, identity, prepared); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `SELECT setval('chunks_seq_id_seq', GREATEST(COALESCE((SELECT MAX(seq_id) FROM chunks), 1), 1), true)`); err != nil {
|
||||
return errors.New("advance target chunk sequence")
|
||||
}
|
||||
@@ -191,18 +187,10 @@ func (importer *PostgresImporter) ReplaceWiki(ctx context.Context, identity Post
|
||||
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 1))`, identity.KnowledgeBaseID); err != nil {
|
||||
return errors.New("lock target wiki knowledge base")
|
||||
}
|
||||
for _, statement := range []clearStatement{
|
||||
{`DELETE FROM wiki_pages WHERE tenant_id = $1 AND knowledge_base_id = $2`, []any{identity.TenantID, identity.KnowledgeBaseID}},
|
||||
{`DELETE FROM wiki_folders WHERE tenant_id = $1 AND knowledge_base_id = $2`, []any{identity.TenantID, identity.KnowledgeBaseID}},
|
||||
} {
|
||||
if _, err := tx.Exec(ctx, statement.sql, statement.args...); err != nil {
|
||||
return errors.New("clear prior target wiki rows")
|
||||
}
|
||||
}
|
||||
for _, table := range tables {
|
||||
if err := insertSnapshotRows(ctx, tx, table); err != nil {
|
||||
return err
|
||||
}
|
||||
// 撤权 / 回滚也走同一套归属判定：只撤本包知识拥有的页面与引用，
|
||||
// 目标库里其他文档和 agent 手写的内容一行不动。
|
||||
if _, err := importer.replaceWiki(ctx, tx, identity, snapshot); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return errors.New("commit PostgreSQL wiki import")
|
||||
|
||||
@@ -0,0 +1,374 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// wikiReplaceScope 描述一次库级 wiki 替换的真实作用域，以及它是否被降级。
|
||||
//
|
||||
// 背景：WeKnora 的 wiki 图谱与页面同表（wiki_pages，节点/边靠 slug + in_links
|
||||
// / out_links 表达），旧的"整库 DELETE + 灌包"在**共享真实知识库**上会把其他
|
||||
// 文档的页面和 agent 手写页一起抹掉（2026-10-08 生产演练的现场）。
|
||||
//
|
||||
// 现在的判定规则：
|
||||
//
|
||||
// - 目标页的 source_refs 形如 "<knowledge_id>|<doc_title>"，只有引用的知识
|
||||
// 全部属于"本包覆盖的知识集合"时，这一页才归本包管（InScope）；
|
||||
// - 引用了包外知识的页是外来页（Foreign），永不删除；
|
||||
// - 完全没有 source_refs/chunk_refs 的页（index/log、agent 手写页）永不删除；
|
||||
// - 外来页与包内页同 slug 时无法安全处理 → 拒绝发布（fail-closed）。
|
||||
//
|
||||
// 目标库没有外来页时（"镜像库 → 原件"前提成立）仍走整库替换，行为与旧实现一致；
|
||||
// 只要出现外来页就降级为"只替换包内页集"，其余页面一行不动。
|
||||
type wikiReplaceScope struct {
|
||||
InScope int // 包内知识拥有、会被删除重建的页
|
||||
Foreign int // 引用了包外知识的页，永不删除
|
||||
ForeignReferencingIn int // 外来页中仍引用本包知识的页（只摘引用，不动正文）
|
||||
Untouched int // 无引用来源的页（index/log/agent），永不触碰
|
||||
SkippedLibraryPages int // 降级时被跳过的库级页（index/log 不能由部分快照重写）
|
||||
Degraded bool // 目标库存在外来页 → 只替换本包页集
|
||||
}
|
||||
|
||||
func (scope wikiReplaceScope) summary() string {
|
||||
return fmt.Sprintf("in_scope=%d foreign=%d foreign_ref_stripped=%d untouched=%d skipped_library=%d degraded=%t",
|
||||
scope.InScope, scope.Foreign, scope.ForeignReferencingIn, scope.Untouched, scope.SkippedLibraryPages, scope.Degraded)
|
||||
}
|
||||
|
||||
// targetWikiRef 是目标库现存 wiki 页面的归属信息。
|
||||
type targetWikiRef struct {
|
||||
ID string
|
||||
Slug string
|
||||
SourceRefs []string
|
||||
ChunkRefs []string
|
||||
}
|
||||
|
||||
const targetWikiRefsSQL = `
|
||||
SELECT id, slug, COALESCE(source_refs, '[]'::jsonb), COALESCE(chunk_refs, '[]'::jsonb)
|
||||
FROM wiki_pages
|
||||
WHERE tenant_id = $1 AND knowledge_base_id = $2 AND deleted_at IS NULL`
|
||||
|
||||
// packageWikiOwners 返回本包 wiki 快照覆盖的知识 id 集合：快照里 knowledges
|
||||
// 行的 id，加上所有页面 source_refs 的归属。
|
||||
//
|
||||
// 镜像库对原件发布时这个集合是整个库（几百上千篇），单文档包发布时它只有一篇
|
||||
// ——所以同一套判定既能保住"整库替换"的镜像语义，又能在共享库上自动收敛。
|
||||
func packageWikiOwners(snapshot PostgresSnapshot) (map[string]bool, error) {
|
||||
owners := map[string]bool{}
|
||||
if tablePresent(snapshot, "knowledges") {
|
||||
for _, raw := range snapshot.Table("knowledges").Rows {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || !validSearchIdentifier(row.ID) {
|
||||
return nil, errors.New("invalid package knowledge row")
|
||||
}
|
||||
owners[row.ID] = true
|
||||
}
|
||||
}
|
||||
if tablePresent(snapshot, "wiki_pages") {
|
||||
for _, raw := range snapshot.Table("wiki_pages").Rows {
|
||||
page, err := decodeWikiRow(raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, reference := range stringList(page["source_refs"]) {
|
||||
if owner := referenceOwner(reference); owner != "" {
|
||||
owners[owner] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(owners) == 0 {
|
||||
return nil, errors.New("package wiki snapshot carries no owning knowledge")
|
||||
}
|
||||
return owners, nil
|
||||
}
|
||||
|
||||
// referenceOwner 取 source_refs 条目的知识 id："<knowledge_id>|<doc_title>"。
|
||||
func referenceOwner(reference string) string {
|
||||
if index := strings.Index(reference, "|"); index > 0 {
|
||||
reference = reference[:index]
|
||||
}
|
||||
if !validSearchIdentifier(reference) {
|
||||
return ""
|
||||
}
|
||||
return reference
|
||||
}
|
||||
|
||||
// classifyTargetWikiPages 判定目标库里每一页的归属（纯函数，便于单测）。
|
||||
// 返回作用域、将被替换的包内 slug，以及"仍引用本包知识的外来页"。
|
||||
func classifyTargetWikiPages(pages []targetWikiRef, owners map[string]bool) (wikiReplaceScope, []string, []targetWikiRef) {
|
||||
scope := wikiReplaceScope{}
|
||||
inScopeSlugs := make([]string, 0, len(pages))
|
||||
foreignReferencingIn := make([]targetWikiRef, 0)
|
||||
for _, page := range pages {
|
||||
if page.Slug == "" || (!validSearchIdentifier(page.ID)) {
|
||||
// 结构异常的页按外来页处理：不删、不改。
|
||||
scope.Foreign++
|
||||
continue
|
||||
}
|
||||
if len(page.SourceRefs) == 0 && len(page.ChunkRefs) == 0 {
|
||||
scope.Untouched++
|
||||
continue
|
||||
}
|
||||
if len(page.SourceRefs) == 0 {
|
||||
// 只有 chunk_refs（chunk UUID 不带知识 id）时无法归属 → 按外来页处理。
|
||||
scope.Foreign++
|
||||
continue
|
||||
}
|
||||
pageOwners := make([]string, 0, len(page.SourceRefs))
|
||||
mine, malformed := 0, false
|
||||
for _, reference := range page.SourceRefs {
|
||||
owner := referenceOwner(reference)
|
||||
if owner == "" {
|
||||
malformed = true
|
||||
break
|
||||
}
|
||||
pageOwners = append(pageOwners, owner)
|
||||
if owners[owner] {
|
||||
mine++
|
||||
}
|
||||
}
|
||||
if malformed {
|
||||
scope.Foreign++
|
||||
continue
|
||||
}
|
||||
if mine == len(pageOwners) {
|
||||
scope.InScope++
|
||||
inScopeSlugs = append(inScopeSlugs, page.Slug)
|
||||
continue
|
||||
}
|
||||
scope.Foreign++
|
||||
if mine > 0 {
|
||||
scope.ForeignReferencingIn++
|
||||
foreignReferencingIn = append(foreignReferencingIn, page)
|
||||
}
|
||||
}
|
||||
scope.Degraded = scope.Foreign > 0
|
||||
return scope, inScopeSlugs, foreignReferencingIn
|
||||
}
|
||||
|
||||
// replaceWiki 是发布 / 撤权 / 回滚共用的 wiki 通道：先判定归属，再按作用域写入。
|
||||
// 必须在已经持有 knowledge base 维度 advisory lock 的事务里调用。
|
||||
func (importer *PostgresImporter) replaceWiki(ctx context.Context, tx pgx.Tx, identity PostgresIdentity, snapshot PostgresSnapshot) (wikiReplaceScope, error) {
|
||||
tables, hasWiki := wikiSnapshotRows(snapshot)
|
||||
if !hasWiki {
|
||||
// 老包 / 未启用 wiki 的库：不动目标 wiki。
|
||||
return wikiReplaceScope{}, nil
|
||||
}
|
||||
pageRows := snapshot.Table("wiki_pages").Rows
|
||||
folderRows := snapshot.Table("wiki_folders").Rows
|
||||
bundleSlugs := make([]string, 0, len(pageRows))
|
||||
for _, raw := range pageRows {
|
||||
var row struct {
|
||||
Slug string `json:"slug"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || row.Slug == "" {
|
||||
return wikiReplaceScope{}, errors.New("invalid package wiki page slug")
|
||||
}
|
||||
bundleSlugs = append(bundleSlugs, row.Slug)
|
||||
}
|
||||
folderIDs := make([]string, 0, len(folderRows))
|
||||
for _, raw := range folderRows {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || row.ID == "" {
|
||||
return wikiReplaceScope{}, errors.New("invalid package wiki folder id")
|
||||
}
|
||||
folderIDs = append(folderIDs, row.ID)
|
||||
}
|
||||
owners, err := packageWikiOwners(snapshot)
|
||||
if err != nil {
|
||||
return wikiReplaceScope{}, err
|
||||
}
|
||||
targetPages, err := loadTargetWikiRefs(ctx, tx, identity)
|
||||
if err != nil {
|
||||
return wikiReplaceScope{}, err
|
||||
}
|
||||
scope, inScopeSlugs, foreignReferencingIn := classifyTargetWikiPages(targetPages, owners)
|
||||
// 外来页与包内会写入的页同 slug：两边正文不同、又都不该被覆盖 → 拒绝发布。
|
||||
bundleRefBearing := make(map[string]bool, len(bundleSlugs))
|
||||
for _, raw := range pageRows {
|
||||
page, err := decodeWikiRow(raw)
|
||||
if err != nil {
|
||||
return scope, err
|
||||
}
|
||||
if len(stringList(page["source_refs"])) == 0 && len(stringList(page["chunk_refs"])) == 0 {
|
||||
continue
|
||||
}
|
||||
slug, _ := page["slug"].(string)
|
||||
bundleRefBearing[slug] = true
|
||||
}
|
||||
if slug, collided := wikiSlugCollision(targetPages, inScopeSlugs, bundleRefBearing); collided {
|
||||
return scope, fmt.Errorf("目标知识库的 wiki 页 %s 由其他文档或人工维护，发布包无法安全覆盖，已拒绝发布（%s）", slug, scope.summary())
|
||||
}
|
||||
if scope.Degraded {
|
||||
// 降级：只摘掉外来页上对本包知识的引用（正文不动，避免破坏库级合成结果）。
|
||||
if err := stripForeignWikiRefs(ctx, tx, identity, foreignReferencingIn, owners, chunkIDSet(snapshot)); err != nil {
|
||||
return scope, err
|
||||
}
|
||||
if len(inScopeSlugs) > 0 {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM wiki_pages
|
||||
WHERE tenant_id = $1 AND knowledge_base_id = $2 AND slug = ANY($3::text[])
|
||||
`, identity.TenantID, identity.KnowledgeBaseID, inScopeSlugs); err != nil {
|
||||
return scope, errors.New("clear in-scope target wiki pages")
|
||||
}
|
||||
}
|
||||
if len(folderIDs) > 0 {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM wiki_folders
|
||||
WHERE tenant_id = $1 AND knowledge_base_id = $2 AND id = ANY($3::text[])
|
||||
`, identity.TenantID, identity.KnowledgeBaseID, folderIDs); err != nil {
|
||||
return scope, errors.New("clear package target wiki folders")
|
||||
}
|
||||
}
|
||||
filteredPages := make([]json.RawMessage, 0, len(pageRows))
|
||||
for _, raw := range pageRows {
|
||||
page, err := decodeWikiRow(raw)
|
||||
if err != nil {
|
||||
return scope, err
|
||||
}
|
||||
if len(stringList(page["source_refs"])) == 0 && len(stringList(page["chunk_refs"])) == 0 {
|
||||
// index/log 是库级聚合页，部分快照重写会毁掉目标库自己的聚合。
|
||||
scope.SkippedLibraryPages++
|
||||
continue
|
||||
}
|
||||
filteredPages = append(filteredPages, raw)
|
||||
}
|
||||
if err := insertSnapshotRows(ctx, tx, newTableSnapshot("wiki_folders", folderRows)); err != nil {
|
||||
return scope, err
|
||||
}
|
||||
if err := insertSnapshotRows(ctx, tx, newTableSnapshot("wiki_pages", filteredPages)); err != nil {
|
||||
return scope, err
|
||||
}
|
||||
log.Printf("release wiki replace scope %s (只替换包内页集，目标库其余页面未改动)", scope.summary())
|
||||
return scope, nil
|
||||
}
|
||||
// 无外来页：镜像库对原件的正常路径，整库替换（与旧行为一致），
|
||||
// 但保留"无引用来源且不在包内"的页面（agent 手写页等）。
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM wiki_pages
|
||||
WHERE tenant_id = $1 AND knowledge_base_id = $2
|
||||
AND ((COALESCE(source_refs, '[]'::jsonb) <> '[]'::jsonb) OR (COALESCE(chunk_refs, '[]'::jsonb) <> '[]'::jsonb)
|
||||
OR slug = ANY($3::text[]))
|
||||
`, identity.TenantID, identity.KnowledgeBaseID, bundleSlugs); err != nil {
|
||||
return scope, errors.New("clear target wiki pages")
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM wiki_folders AS folder
|
||||
WHERE folder.tenant_id = $1 AND folder.knowledge_base_id = $2
|
||||
AND (folder.id = ANY($3::text[])
|
||||
OR NOT EXISTS (SELECT 1 FROM wiki_pages AS page
|
||||
WHERE page.knowledge_base_id = folder.knowledge_base_id
|
||||
AND page.folder_id = folder.id AND page.deleted_at IS NULL))
|
||||
`, identity.TenantID, identity.KnowledgeBaseID, folderIDs); err != nil {
|
||||
return scope, errors.New("clear target wiki folders")
|
||||
}
|
||||
for _, table := range tables {
|
||||
if err := insertSnapshotRows(ctx, tx, table); err != nil {
|
||||
return scope, err
|
||||
}
|
||||
}
|
||||
log.Printf("release wiki replace scope %s", scope.summary())
|
||||
return scope, nil
|
||||
}
|
||||
|
||||
func loadTargetWikiRefs(ctx context.Context, tx pgx.Tx, identity PostgresIdentity) ([]targetWikiRef, error) {
|
||||
rows, err := tx.Query(ctx, targetWikiRefsSQL, identity.TenantID, identity.KnowledgeBaseID)
|
||||
if err != nil {
|
||||
return nil, errors.New("read target wiki page ownership")
|
||||
}
|
||||
defer rows.Close()
|
||||
pages := make([]targetWikiRef, 0)
|
||||
for rows.Next() {
|
||||
var page targetWikiRef
|
||||
var sourceRefs, chunkRefs []byte
|
||||
if err := rows.Scan(&page.ID, &page.Slug, &sourceRefs, &chunkRefs); err != nil {
|
||||
return nil, errors.New("read target wiki page ownership")
|
||||
}
|
||||
if json.Unmarshal(sourceRefs, &page.SourceRefs) != nil || json.Unmarshal(chunkRefs, &page.ChunkRefs) != nil {
|
||||
// 非数组结构（异常行）按"无法归属"处理：不删、不改。
|
||||
page.SourceRefs = nil
|
||||
page.ChunkRefs = []string{""}
|
||||
}
|
||||
pages = append(pages, page)
|
||||
}
|
||||
if rows.Err() != nil {
|
||||
return nil, errors.New("read target wiki page ownership")
|
||||
}
|
||||
return pages, nil
|
||||
}
|
||||
|
||||
// stripForeignWikiRefs 在外来页上只摘掉本包知识的引用，正文一行不改。
|
||||
func stripForeignWikiRefs(ctx context.Context, tx pgx.Tx, identity PostgresIdentity, pages []targetWikiRef, owners map[string]bool, removedChunks map[string]bool) error {
|
||||
for _, page := range pages {
|
||||
keptSources := make([]string, 0, len(page.SourceRefs))
|
||||
for _, reference := range page.SourceRefs {
|
||||
if !owners[referenceOwner(reference)] {
|
||||
keptSources = append(keptSources, reference)
|
||||
}
|
||||
}
|
||||
keptChunks := make([]string, 0, len(page.ChunkRefs))
|
||||
for _, reference := range page.ChunkRefs {
|
||||
if !removedChunks[reference] {
|
||||
keptChunks = append(keptChunks, reference)
|
||||
}
|
||||
}
|
||||
if len(keptSources) == len(page.SourceRefs) && len(keptChunks) == len(page.ChunkRefs) {
|
||||
continue
|
||||
}
|
||||
encodedSources, _ := json.Marshal(keptSources)
|
||||
encodedChunks, _ := json.Marshal(keptChunks)
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE wiki_pages SET source_refs = $3::jsonb, chunk_refs = $4::jsonb, updated_at = NOW()
|
||||
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
||||
`, page.ID, identity.TenantID, string(encodedSources), string(encodedChunks)); err != nil {
|
||||
return errors.New("strip package wiki references from shared target page")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// wikiSlugCollision 找"包内会写入的页 slug 已被目标库外来页占用"的第一个 slug。
|
||||
// 命中时两侧正文不同、又都不该被覆盖，只能拒绝发布（fail-closed）。
|
||||
//
|
||||
// bundleRefBearing 是包内**携带引用**的页 slug：只有这些页在降级路径上会真正
|
||||
// 写入；包内的库级页（index/log）会被跳过，因此不构成占用。
|
||||
func wikiSlugCollision(pages []targetWikiRef, inScopeSlugs []string, bundleRefBearing map[string]bool) (string, bool) {
|
||||
inScope := make(map[string]bool, len(inScopeSlugs))
|
||||
for _, slug := range inScopeSlugs {
|
||||
inScope[slug] = true
|
||||
}
|
||||
for _, page := range pages {
|
||||
if !bundleRefBearing[page.Slug] || inScope[page.Slug] {
|
||||
continue
|
||||
}
|
||||
return page.Slug, true
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
func chunkIDSet(snapshot PostgresSnapshot) map[string]bool {
|
||||
ids := map[string]bool{}
|
||||
if !tablePresent(snapshot, "chunks") {
|
||||
return ids
|
||||
}
|
||||
for _, raw := range snapshot.Table("chunks").Rows {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) == nil && row.ID != "" {
|
||||
ids[row.ID] = true
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
@@ -0,0 +1,453 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// 这组用例在真实的 WeKnora v0.7.1 库上验证"共享知识库只动本包页集"的语义：
|
||||
// 源库（测试库）导出 → 重映射到目标库 → 目标库上带外来页/agent 页导入。
|
||||
//
|
||||
// 只使用本次运行新建的临时知识库（id 带随机后缀），结束时按 id 精确清理。
|
||||
const wikiGateTenant = 10000
|
||||
|
||||
type wikiGateFixture struct {
|
||||
pool *pgxpool.Pool
|
||||
ctx context.Context
|
||||
suffix string
|
||||
sourceK string
|
||||
targetK string
|
||||
ours string
|
||||
theirs string
|
||||
}
|
||||
|
||||
func newWikiGateFixture(t *testing.T) (*wikiGateFixture, bool) {
|
||||
t.Helper()
|
||||
url := os.Getenv("TEST_WEKNORA_V071_TARGET_DATABASE_URL")
|
||||
if url == "" {
|
||||
url = os.Getenv("TEST_WEKNORA_V071_DATABASE_URL")
|
||||
}
|
||||
if url == "" {
|
||||
return nil, false
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, url)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buffer := make([]byte, 4)
|
||||
if _, err := readRandom(buffer); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
suffix := hex.EncodeToString(buffer)
|
||||
fixture := &wikiGateFixture{
|
||||
pool: pool, ctx: ctx, suffix: suffix,
|
||||
sourceK: "kb-wiki-src-" + suffix, targetK: "kb-wiki-tgt-" + suffix,
|
||||
ours: "knowledge-wiki-src-" + suffix, theirs: "knowledge-wiki-other-" + suffix,
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
fixture.cleanup(t)
|
||||
pool.Close()
|
||||
})
|
||||
fixture.cleanup(t)
|
||||
return fixture, true
|
||||
}
|
||||
|
||||
func readRandom(buffer []byte) (int, error) {
|
||||
file, err := os.Open("/dev/urandom")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer file.Close()
|
||||
return file.Read(buffer)
|
||||
}
|
||||
|
||||
func (fixture *wikiGateFixture) exec(t *testing.T, sql string, arguments ...any) {
|
||||
t.Helper()
|
||||
if _, err := fixture.pool.Exec(fixture.ctx, sql, arguments...); err != nil {
|
||||
t.Fatalf("exec %s: %v", firstLineSQL(sql), err)
|
||||
}
|
||||
}
|
||||
|
||||
func firstLineSQL(sql string) string {
|
||||
for index, char := range sql {
|
||||
if char == '\n' {
|
||||
return sql[:index]
|
||||
}
|
||||
}
|
||||
return sql
|
||||
}
|
||||
|
||||
func (fixture *wikiGateFixture) cleanup(t *testing.T) {
|
||||
t.Helper()
|
||||
for _, table := range []string{"wiki_pages", "wiki_folders", "chunks", "knowledges"} {
|
||||
if _, err := fixture.pool.Exec(fixture.ctx,
|
||||
fmt.Sprintf("DELETE FROM %s WHERE knowledge_base_id = ANY($1::text[])", table),
|
||||
[]string{fixture.sourceK, fixture.targetK}); err != nil {
|
||||
t.Logf("cleanup %s: %v", table, err)
|
||||
}
|
||||
}
|
||||
if _, err := fixture.pool.Exec(fixture.ctx,
|
||||
`DELETE FROM knowledge_bases WHERE id = ANY($1::text[])`, []string{fixture.sourceK, fixture.targetK}); err != nil {
|
||||
t.Logf("cleanup knowledge_bases: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// templateRowID 找一个可复制的真实行，避免手写全部 NOT NULL 列。
|
||||
func (fixture *wikiGateFixture) templateRowID(t *testing.T, table string) string {
|
||||
t.Helper()
|
||||
column := "knowledge_base_id"
|
||||
if table == "knowledge_bases" {
|
||||
column = "id"
|
||||
}
|
||||
var id string
|
||||
if err := fixture.pool.QueryRow(fixture.ctx,
|
||||
fmt.Sprintf("SELECT id FROM %s WHERE %s <> ALL($1::text[]) AND deleted_at IS NULL LIMIT 1", table, column),
|
||||
[]string{fixture.sourceK, fixture.targetK}).Scan(&id); err != nil {
|
||||
t.Skipf("no template row in %s: %v", table, err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func (fixture *wikiGateFixture) cloneRow(t *testing.T, table string, templateID string, overrides map[string]any) {
|
||||
t.Helper()
|
||||
encoded, err := json.Marshal(overrides)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fixture.exec(t, fmt.Sprintf(`
|
||||
INSERT INTO %s
|
||||
SELECT (jsonb_populate_record(NULL::%s, to_jsonb(template) || $1::jsonb)).*
|
||||
FROM %s AS template WHERE template.id = $2
|
||||
`, table, table, table), string(encoded), templateID)
|
||||
}
|
||||
|
||||
// seedSourceKnowledgeBase 建一个"测试库"：启用 wiki、一篇完成解析的文档 + 一个 chunk
|
||||
// + 我们自己的页与被撤权页。
|
||||
func (fixture *wikiGateFixture) seedSourceKnowledgeBase(t *testing.T) {
|
||||
t.Helper()
|
||||
fixture.cloneRow(t, "knowledge_bases", fixture.templateRowID(t, "knowledge_bases"), map[string]any{
|
||||
"id": fixture.sourceK, "tenant_id": wikiGateTenant, "name": "wiki-gate-source-" + fixture.suffix,
|
||||
"indexing_strategy": map[string]any{"wiki_enabled": true},
|
||||
})
|
||||
fixture.cloneRow(t, "knowledges", fixture.templateRowID(t, "knowledges"), map[string]any{
|
||||
"id": fixture.ours, "tenant_id": wikiGateTenant, "knowledge_base_id": fixture.sourceK,
|
||||
"title": "WikiGate Ours.docx", "parse_status": "completed", "enable_status": "enabled",
|
||||
"pending_subtasks_count": 0, "deleted_at": nil,
|
||||
})
|
||||
// chunks.seq_id 有唯一索引：克隆必须换成新序号，不能带模板的值。
|
||||
var chunkSeq int64
|
||||
if err := fixture.pool.QueryRow(fixture.ctx, `SELECT nextval('chunks_seq_id_seq')`).Scan(&chunkSeq); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fixture.cloneRow(t, "chunks", fixture.templateRowID(t, "chunks"), map[string]any{
|
||||
"id": "chunk-wiki-src-" + fixture.suffix, "tenant_id": wikiGateTenant,
|
||||
"knowledge_id": fixture.ours, "knowledge_base_id": fixture.sourceK,
|
||||
"is_enabled": true, "content": "WikiGate source chunk", "deleted_at": nil,
|
||||
"seq_id": chunkSeq,
|
||||
})
|
||||
fixture.cloneRow(t, "wiki_folders", fixture.templateRowID(t, "wiki_folders"), map[string]any{
|
||||
"id": "folder-wiki-src-" + fixture.suffix, "tenant_id": wikiGateTenant,
|
||||
"knowledge_base_id": fixture.sourceK, "name": "WikiGate " + fixture.suffix,
|
||||
"parent_id": "", "deleted_at": nil,
|
||||
})
|
||||
for index, slug := range []string{"summary/" + fixture.ours, "index"} {
|
||||
refs := []string{fixture.ours + "|WikiGate Ours.docx"}
|
||||
chunkRefs := []string{"chunk-wiki-src-" + fixture.suffix}
|
||||
if slug == "index" {
|
||||
refs, chunkRefs = []string{}, []string{}
|
||||
}
|
||||
fixture.cloneRow(t, "wiki_pages", fixture.templateRowID(t, "wiki_pages"), map[string]any{
|
||||
"id": fmt.Sprintf("page-wiki-src-%s-%d", fixture.suffix, index), "tenant_id": wikiGateTenant,
|
||||
"knowledge_base_id": fixture.sourceK, "slug": slug, "title": "WikiGate " + slug,
|
||||
"content": "package content for " + slug, "source_refs": refs, "chunk_refs": chunkRefs,
|
||||
"folder_id": "folder-wiki-src-" + fixture.suffix, "in_links": []string{}, "out_links": []string{},
|
||||
"deleted_at": nil,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// seedTargetKnowledgeBase 建一个"共享真实库"：别人的页、我们与别人共享的页、agent 手写页、
|
||||
// 以及库级 index。
|
||||
func (fixture *wikiGateFixture) seedTargetKnowledgeBase(t *testing.T) {
|
||||
t.Helper()
|
||||
fixture.cloneRow(t, "knowledge_bases", fixture.templateRowID(t, "knowledge_bases"), map[string]any{
|
||||
"id": fixture.targetK, "tenant_id": wikiGateTenant, "name": "wiki-gate-target-" + fixture.suffix,
|
||||
"indexing_strategy": map[string]any{"wiki_enabled": true},
|
||||
})
|
||||
fixture.cloneRow(t, "wiki_folders", fixture.templateRowID(t, "wiki_folders"), map[string]any{
|
||||
"id": "folder-wiki-theirs-" + fixture.suffix, "tenant_id": wikiGateTenant,
|
||||
"knowledge_base_id": fixture.targetK, "name": "Theirs " + fixture.suffix,
|
||||
"parent_id": "", "deleted_at": nil,
|
||||
})
|
||||
pages := []struct {
|
||||
id, slug, content string
|
||||
refs []string
|
||||
}{
|
||||
{"page-theirs-" + fixture.suffix, "summary/" + fixture.theirs, "theirs content", []string{fixture.theirs + "|Theirs.xlsx"}},
|
||||
{"page-shared-" + fixture.suffix, "entity/shared", "merged content from both documents", []string{fixture.ours + "|WikiGate Ours.docx", fixture.theirs + "|Theirs.xlsx"}},
|
||||
{"page-agent-" + fixture.suffix, "concept/hand-made", "agent written page", []string{}},
|
||||
{"page-index-" + fixture.suffix, "index", "target library index", []string{}},
|
||||
}
|
||||
for _, page := range pages {
|
||||
fixture.cloneRow(t, "wiki_pages", fixture.templateRowID(t, "wiki_pages"), map[string]any{
|
||||
"id": page.id, "tenant_id": wikiGateTenant, "knowledge_base_id": fixture.targetK,
|
||||
"slug": page.slug, "title": "Theirs " + page.slug, "content": page.content,
|
||||
"source_refs": page.refs, "chunk_refs": []string{},
|
||||
"folder_id": "folder-wiki-theirs-" + fixture.suffix, "in_links": []string{}, "out_links": []string{},
|
||||
"deleted_at": nil,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// exportSourceSnapshot 用真实导出器取包，再把库 id 重映射成目标库（与发布链一致）。
|
||||
func (fixture *wikiGateFixture) exportSourceSnapshot(t *testing.T) PostgresSnapshot {
|
||||
t.Helper()
|
||||
exporter, err := NewPostgresExporter(fixture.pool)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
identity := PostgresIdentity{TenantID: wikiGateTenant, KnowledgeBaseID: fixture.sourceK, KnowledgeID: fixture.ours}
|
||||
snapshot, err := exporter.Export(fixture.ctx, identity)
|
||||
if err != nil {
|
||||
t.Fatalf("export source snapshot: %v", err)
|
||||
}
|
||||
// 与文档导出器一致：库级 wiki 快照追加进同一个发布单元。
|
||||
wikiTables, err := exporter.ExportWiki(fixture.ctx, identity, "")
|
||||
if err != nil {
|
||||
t.Fatalf("export source wiki snapshot: %v", err)
|
||||
}
|
||||
snapshot.Tables = append(snapshot.Tables, wikiTables...)
|
||||
remapped, err := remapSnapshotKnowledgeBase(snapshot, fixture.sourceK, fixture.targetK)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 真实部署里源库与目标库在不同环境；本地跑同一个库时，先把源库行清掉，
|
||||
// 否则目标端"同一 knowledge id 不能属于别的库"的守卫会正确拦住我们。
|
||||
fixture.dropSourceRows(t)
|
||||
return remapped
|
||||
}
|
||||
|
||||
func (fixture *wikiGateFixture) dropSourceRows(t *testing.T) {
|
||||
t.Helper()
|
||||
for _, table := range []string{"wiki_pages", "wiki_folders", "chunks", "knowledges"} {
|
||||
fixture.exec(t, fmt.Sprintf("DELETE FROM %s WHERE knowledge_base_id = $1", table), fixture.sourceK)
|
||||
}
|
||||
fixture.exec(t, `DELETE FROM knowledge_bases WHERE id = $1`, fixture.sourceK)
|
||||
}
|
||||
|
||||
func remapSnapshotKnowledgeBase(snapshot PostgresSnapshot, sourceID, targetID string) (PostgresSnapshot, error) {
|
||||
result := cloneSnapshot(snapshot)
|
||||
remappable := map[string]bool{}
|
||||
for _, name := range remappableTables {
|
||||
remappable[name] = true
|
||||
}
|
||||
for index, table := range result.Tables {
|
||||
if !remappable[table.Name] {
|
||||
// 与发布链一致：只有携带库身份的表演进重映射。
|
||||
continue
|
||||
}
|
||||
rows := make([]json.RawMessage, 0, len(table.Rows))
|
||||
for _, raw := range table.Rows {
|
||||
rewritten, err := remapKnowledgeBaseField(raw, sourceID, targetID)
|
||||
if err != nil {
|
||||
return PostgresSnapshot{}, err
|
||||
}
|
||||
rows = append(rows, rewritten)
|
||||
}
|
||||
result.Tables[index] = newTableSnapshot(table.Name, rows)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (fixture *wikiGateFixture) pageState(t *testing.T, slug string) (content string, sourceRefs []string, present bool) {
|
||||
t.Helper()
|
||||
var rawContent, refs []byte
|
||||
err := fixture.pool.QueryRow(fixture.ctx, `
|
||||
SELECT content, COALESCE(source_refs, '[]'::jsonb)
|
||||
FROM wiki_pages
|
||||
WHERE tenant_id = $1 AND knowledge_base_id = $2 AND slug = $3 AND deleted_at IS NULL
|
||||
`, wikiGateTenant, fixture.targetK, slug).Scan(&rawContent, &refs)
|
||||
if err != nil {
|
||||
return "", nil, false
|
||||
}
|
||||
if json.Unmarshal(refs, &sourceRefs) != nil {
|
||||
t.Fatalf("decode refs for %s", slug)
|
||||
}
|
||||
return string(rawContent), sourceRefs, true
|
||||
}
|
||||
|
||||
func TestWikiReplaceOnSharedKnowledgeBaseKeepsForeignPages(t *testing.T) {
|
||||
fixture, ok := newWikiGateFixture(t)
|
||||
if !ok {
|
||||
t.Skip("v0.7.1 integration database URL is not set")
|
||||
}
|
||||
fixture.seedSourceKnowledgeBase(t)
|
||||
fixture.seedTargetKnowledgeBase(t)
|
||||
snapshot := fixture.exportSourceSnapshot(t)
|
||||
importer, err := NewPostgresImporter(fixture.pool, PostgresImporterConfig{
|
||||
TargetStorageBackendID: "6129cfd1-a26c-41dd-88cb-e6a800daa3bb", TargetProvider: "oss",
|
||||
TargetBucket: "weknora-gate-test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := importer.Replace(fixture.ctx, PostgresIdentity{
|
||||
TenantID: wikiGateTenant, KnowledgeBaseID: fixture.targetK, KnowledgeID: fixture.ours,
|
||||
}, snapshot); err != nil {
|
||||
t.Fatalf("replace on shared knowledge base: %v", err)
|
||||
}
|
||||
|
||||
// 包内页按快照重建。
|
||||
content, refs, present := fixture.pageState(t, "summary/"+fixture.ours)
|
||||
if !present || content != "package content for summary/"+fixture.ours {
|
||||
t.Fatalf("in-scope page not replaced: present=%t content=%q", present, content)
|
||||
}
|
||||
if len(refs) != 1 || !referencesKnowledge(refs[0], fixture.ours) {
|
||||
t.Fatalf("in-scope page refs=%v", refs)
|
||||
}
|
||||
// 别人的页一行不动。
|
||||
if content, _, present := fixture.pageState(t, "summary/"+fixture.theirs); !present || content != "theirs content" {
|
||||
t.Fatalf("foreign page must survive: present=%t content=%q", present, content)
|
||||
}
|
||||
// agent 手写页一行不动。
|
||||
if content, _, present := fixture.pageState(t, "concept/hand-made"); !present || content != "agent written page" {
|
||||
t.Fatalf("agent page must survive: present=%t content=%q", present, content)
|
||||
}
|
||||
// 库级 index 不被部分快照重写。
|
||||
if content, _, present := fixture.pageState(t, "index"); !present || content != "target library index" {
|
||||
t.Fatalf("library index must survive: present=%t content=%q", present, content)
|
||||
}
|
||||
// 共享页保留正文，只摘掉本包知识的引用。
|
||||
content, refs, present = fixture.pageState(t, "entity/shared")
|
||||
if !present || content != "merged content from both documents" {
|
||||
t.Fatalf("shared page content must be untouched: present=%t content=%q", present, content)
|
||||
}
|
||||
for _, reference := range refs {
|
||||
if referencesKnowledge(reference, fixture.ours) {
|
||||
t.Fatalf("shared page still references the withdrawn/updated package document: %v", refs)
|
||||
}
|
||||
}
|
||||
if len(refs) == 0 {
|
||||
t.Fatal("shared page must keep the other document's reference")
|
||||
}
|
||||
// 库里的总页数：包内 2 页（summary + 不写 index）+ 别人的 4 页。
|
||||
var total int
|
||||
if err := fixture.pool.QueryRow(fixture.ctx,
|
||||
`SELECT COUNT(*)::INT FROM wiki_pages WHERE tenant_id = $1 AND knowledge_base_id = $2 AND deleted_at IS NULL`,
|
||||
wikiGateTenant, fixture.targetK).Scan(&total); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if total != 5 {
|
||||
t.Fatalf("target wiki page count=%d, want 5 (2 package + 3 foreign/untouched)", total)
|
||||
}
|
||||
}
|
||||
|
||||
// 包内页 slug 已被外来页占用时必须拒绝发布，且一行都不能改。
|
||||
func TestWikiReplaceRefusesWhenPackageSlugBelongsToForeignPage(t *testing.T) {
|
||||
fixture, ok := newWikiGateFixture(t)
|
||||
if !ok {
|
||||
t.Skip("v0.7.1 integration database URL is not set")
|
||||
}
|
||||
fixture.seedSourceKnowledgeBase(t)
|
||||
fixture.seedTargetKnowledgeBase(t)
|
||||
snapshot := fixture.exportSourceSnapshot(t)
|
||||
// 把包里 summary 页的 slug 抢到外来页名下：目标库该 slug 归属别人。
|
||||
fixture.cloneRow(t, "wiki_pages", fixture.templateRowID(t, "wiki_pages"), map[string]any{
|
||||
"id": "page-contested-" + fixture.suffix, "tenant_id": wikiGateTenant,
|
||||
"knowledge_base_id": fixture.targetK, "slug": "summary/" + fixture.ours,
|
||||
"title": "Theirs contested", "content": "theirs content at the package slug",
|
||||
"source_refs": []string{fixture.theirs + "|Theirs.xlsx"}, "chunk_refs": []string{},
|
||||
"folder_id": "folder-wiki-theirs-" + fixture.suffix, "in_links": []string{}, "out_links": []string{},
|
||||
"deleted_at": nil,
|
||||
})
|
||||
importer, err := NewPostgresImporter(fixture.pool, PostgresImporterConfig{
|
||||
TargetStorageBackendID: "6129cfd1-a26c-41dd-88cb-e6a800daa3bb", TargetProvider: "oss",
|
||||
TargetBucket: "weknora-gate-test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = importer.Replace(fixture.ctx, PostgresIdentity{
|
||||
TenantID: wikiGateTenant, KnowledgeBaseID: fixture.targetK, KnowledgeID: fixture.ours,
|
||||
}, snapshot)
|
||||
if err == nil {
|
||||
t.Fatal("expected a fail-closed refusal when a package slug is owned by another document")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "已拒绝发布") {
|
||||
t.Fatalf("expected the wiki ownership refusal, got: %v", err)
|
||||
}
|
||||
var total int
|
||||
if queryErr := fixture.pool.QueryRow(fixture.ctx,
|
||||
`SELECT COUNT(*)::INT FROM wiki_pages WHERE tenant_id = $1 AND knowledge_base_id = $2 AND deleted_at IS NULL`,
|
||||
wikiGateTenant, fixture.targetK).Scan(&total); queryErr != nil {
|
||||
t.Fatal(queryErr)
|
||||
}
|
||||
if total != 5 {
|
||||
t.Fatalf("refused release must not change target wiki: page count=%d", total)
|
||||
}
|
||||
if content, _, present := fixture.pageState(t, "summary/"+fixture.ours); !present || content != "theirs content at the package slug" {
|
||||
t.Fatalf("contested foreign content must be intact after refusal: %t %q", present, content)
|
||||
}
|
||||
}
|
||||
|
||||
// 镜像库对原件（目标库没有外来页）时必须保持旧的整库替换语义：
|
||||
// 包里的库级页（index）也要写进去，不能因为降级保护而漏掉。
|
||||
func TestWikiReplaceOnFaithfulMirrorReplacesWholeLibrary(t *testing.T) {
|
||||
fixture, ok := newWikiGateFixture(t)
|
||||
if !ok {
|
||||
t.Skip("v0.7.1 integration database URL is not set")
|
||||
}
|
||||
fixture.seedSourceKnowledgeBase(t)
|
||||
fixture.cloneRow(t, "knowledge_bases", fixture.templateRowID(t, "knowledge_bases"), map[string]any{
|
||||
"id": fixture.targetK, "tenant_id": wikiGateTenant, "name": "wiki-gate-mirror-" + fixture.suffix,
|
||||
"indexing_strategy": map[string]any{"wiki_enabled": true},
|
||||
})
|
||||
fixture.cloneRow(t, "wiki_pages", fixture.templateRowID(t, "wiki_pages"), map[string]any{
|
||||
"id": "page-mirror-" + fixture.suffix, "tenant_id": wikiGateTenant, "knowledge_base_id": fixture.targetK,
|
||||
"slug": "summary/" + fixture.ours, "title": "mirror page", "content": "stale mirror content",
|
||||
"source_refs": []string{fixture.ours + "|WikiGate Ours.docx"}, "chunk_refs": []string{},
|
||||
"folder_id": "", "in_links": []string{}, "out_links": []string{}, "deleted_at": nil,
|
||||
})
|
||||
fixture.cloneRow(t, "wiki_pages", fixture.templateRowID(t, "wiki_pages"), map[string]any{
|
||||
"id": "page-mirror-index-" + fixture.suffix, "tenant_id": wikiGateTenant, "knowledge_base_id": fixture.targetK,
|
||||
"slug": "index", "title": "mirror index", "content": "stale mirror index",
|
||||
"source_refs": []string{}, "chunk_refs": []string{},
|
||||
"folder_id": "", "in_links": []string{}, "out_links": []string{}, "deleted_at": nil,
|
||||
})
|
||||
snapshot := fixture.exportSourceSnapshot(t)
|
||||
importer, err := NewPostgresImporter(fixture.pool, PostgresImporterConfig{
|
||||
TargetStorageBackendID: "6129cfd1-a26c-41dd-88cb-e6a800daa3bb", TargetProvider: "oss",
|
||||
TargetBucket: "weknora-gate-test",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := importer.Replace(fixture.ctx, PostgresIdentity{
|
||||
TenantID: wikiGateTenant, KnowledgeBaseID: fixture.targetK, KnowledgeID: fixture.ours,
|
||||
}, snapshot); err != nil {
|
||||
t.Fatalf("replace on faithful mirror: %v", err)
|
||||
}
|
||||
if content, _, present := fixture.pageState(t, "summary/"+fixture.ours); !present || content != "package content for summary/"+fixture.ours {
|
||||
t.Fatalf("mirror page not replaced: present=%t content=%q", present, content)
|
||||
}
|
||||
if content, _, present := fixture.pageState(t, "index"); !present || content != "package content for index" {
|
||||
t.Fatalf("library page must be written on the mirror path: present=%t content=%q", present, content)
|
||||
}
|
||||
var total int
|
||||
if err := fixture.pool.QueryRow(fixture.ctx,
|
||||
`SELECT COUNT(*)::INT FROM wiki_pages WHERE tenant_id = $1 AND knowledge_base_id = $2 AND deleted_at IS NULL`,
|
||||
wikiGateTenant, fixture.targetK).Scan(&total); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if total != 2 {
|
||||
t.Fatalf("mirror replace page count=%d, want 2", total)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func wikiPageRefJSON(t *testing.T, fields map[string]any) json.RawMessage {
|
||||
t.Helper()
|
||||
encoded, err := json.Marshal(fields)
|
||||
if err != nil {
|
||||
t.Fatalf("encode wiki page row: %v", err)
|
||||
}
|
||||
return encoded
|
||||
}
|
||||
|
||||
func contains(values []string, want string) bool {
|
||||
for _, value := range values {
|
||||
if value == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 镜像库对原件：目标库每一页的来源都在包内 → 允许整库替换（旧行为保持不变）。
|
||||
func TestClassifyTargetWikiPagesAllowsLibraryReplaceOnFaithfulMirror(t *testing.T) {
|
||||
owners := map[string]bool{"knowledge-a": true, "knowledge-b": true}
|
||||
pages := []targetWikiRef{
|
||||
{ID: "p1", Slug: "summary/a", SourceRefs: []string{"knowledge-a|A.docx"}},
|
||||
{ID: "p2", Slug: "entity/shared", SourceRefs: []string{"knowledge-a|A.docx", "knowledge-b|B.docx"}},
|
||||
{ID: "p3", Slug: "index", SourceRefs: nil, ChunkRefs: nil},
|
||||
}
|
||||
scope, inScope, foreign := classifyTargetWikiPages(pages, owners)
|
||||
if scope.Degraded || scope.Foreign != 0 {
|
||||
t.Fatalf("mirror must not degrade: %+v", scope)
|
||||
}
|
||||
if scope.InScope != 2 || scope.Untouched != 1 {
|
||||
t.Fatalf("scope=%+v", scope)
|
||||
}
|
||||
if !contains(inScope, "summary/a") || !contains(inScope, "entity/shared") {
|
||||
t.Fatalf("in-scope slugs=%v", inScope)
|
||||
}
|
||||
if len(foreign) != 0 {
|
||||
t.Fatalf("foreign=%v", foreign)
|
||||
}
|
||||
}
|
||||
|
||||
// 共享真实库：目标库存在别的文档的页 → 降级，只换本包页集，其余绝不触碰。
|
||||
func TestClassifyTargetWikiPagesDegradesOnSharedKnowledgeBase(t *testing.T) {
|
||||
owners := map[string]bool{"knowledge-ours": true}
|
||||
pages := []targetWikiRef{
|
||||
{ID: "p1", Slug: "summary/ours", SourceRefs: []string{"knowledge-ours|Ours.docx"}},
|
||||
{ID: "p2", Slug: "entity/shared", SourceRefs: []string{"knowledge-ours|Ours.docx", "knowledge-theirs|Theirs.xlsx"}},
|
||||
{ID: "p3", Slug: "concept/agent-made", SourceRefs: nil, ChunkRefs: nil},
|
||||
{ID: "p4", Slug: "summary/theirs", SourceRefs: []string{"knowledge-theirs|Theirs.xlsx"}},
|
||||
{ID: "p5", Slug: "orphan/chunk-only", SourceRefs: nil, ChunkRefs: []string{"chunk-1"}},
|
||||
}
|
||||
scope, inScope, foreign := classifyTargetWikiPages(pages, owners)
|
||||
if !scope.Degraded {
|
||||
t.Fatalf("shared knowledge base must degrade: %+v", scope)
|
||||
}
|
||||
if scope.InScope != 1 || scope.Foreign != 3 || scope.Untouched != 1 {
|
||||
t.Fatalf("scope=%+v", scope)
|
||||
}
|
||||
if scope.ForeignReferencingIn != 1 || len(foreign) != 1 || foreign[0].Slug != "entity/shared" {
|
||||
t.Fatalf("foreign referencing in-scope must be reported for ref stripping: %+v %v", scope, foreign)
|
||||
}
|
||||
if len(inScope) != 1 || inScope[0] != "summary/ours" {
|
||||
t.Fatalf("in-scope slugs=%v", inScope)
|
||||
}
|
||||
}
|
||||
|
||||
// 结构异常（引用带坏 id / 非数组）的页按外来页处理：不删、不改。
|
||||
func TestClassifyTargetWikiPagesTreatsMalformedRowsAsForeign(t *testing.T) {
|
||||
owners := map[string]bool{"knowledge-ours": true}
|
||||
pages := []targetWikiRef{
|
||||
{ID: "p1", Slug: "summary/ours", SourceRefs: []string{"knowledge-ours|Ours.docx"}},
|
||||
{ID: "p2", Slug: "broken/ref", SourceRefs: []string{"not a valid id|Doc"}},
|
||||
}
|
||||
scope, inScope, _ := classifyTargetWikiPages(pages, owners)
|
||||
if scope.InScope != 1 || scope.Foreign != 1 || !scope.Degraded {
|
||||
t.Fatalf("scope=%+v", scope)
|
||||
}
|
||||
if len(inScope) != 1 {
|
||||
t.Fatalf("in-scope slugs=%v", inScope)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPackageWikiOwnersUnionsKnowledgesAndPageReferences(t *testing.T) {
|
||||
snapshot := PostgresSnapshot{Tables: []TableSnapshot{
|
||||
newTableSnapshot("knowledges", []json.RawMessage{wikiPageRefJSON(t, map[string]any{"id": "knowledge-ours"})}),
|
||||
newTableSnapshot("wiki_pages", []json.RawMessage{
|
||||
wikiPageRefJSON(t, map[string]any{"id": "p1", "slug": "summary/ours", "source_refs": []string{"knowledge-ours|Ours.docx", "knowledge-other|Other.pdf"}}),
|
||||
wikiPageRefJSON(t, map[string]any{"id": "p2", "slug": "index", "source_refs": []string{}}),
|
||||
}),
|
||||
}}
|
||||
owners, err := packageWikiOwners(snapshot)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !owners["knowledge-ours"] || !owners["knowledge-other"] {
|
||||
t.Fatalf("owners=%v", owners)
|
||||
}
|
||||
empty := PostgresSnapshot{Tables: []TableSnapshot{
|
||||
newTableSnapshot("knowledges", []json.RawMessage{wikiPageRefJSON(t, map[string]any{"id": "knowledge-ours"})}),
|
||||
// 只有空 wiki 快照时集合仍来自 knowledges 行。
|
||||
newTableSnapshot("wiki_pages", []json.RawMessage{wikiPageRefJSON(t, map[string]any{"id": "p1", "slug": "index", "source_refs": []string{}})}),
|
||||
}}
|
||||
if _, err := packageWikiOwners(empty); err != nil {
|
||||
t.Fatalf("knowledge rows alone must define ownership: %v", err)
|
||||
}
|
||||
if _, err := packageWikiOwners(PostgresSnapshot{}); err == nil {
|
||||
t.Fatal("empty package must be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReferenceOwnerParsesKnowledgeID(t *testing.T) {
|
||||
if got := referenceOwner("knowledge-abc|手册.pdf"); got != "knowledge-abc" {
|
||||
t.Fatalf("owner=%q", got)
|
||||
}
|
||||
if got := referenceOwner("knowledge-abc"); got != "knowledge-abc" {
|
||||
t.Fatalf("owner=%q", got)
|
||||
}
|
||||
if got := referenceOwner("not a valid id|Doc"); got != "" {
|
||||
t.Fatalf("malformed reference must not yield an owner: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 包内页 slug 已被外来页占用 → 必须拒绝发布。
|
||||
func TestWikiSlugCollisionDetectsForeignSlug(t *testing.T) {
|
||||
pages := []targetWikiRef{
|
||||
{ID: "p1", Slug: "summary/ours", SourceRefs: []string{"knowledge-ours|Ours.docx"}},
|
||||
{ID: "p2", Slug: "entity/contested", SourceRefs: []string{"knowledge-theirs|Theirs.xlsx"}},
|
||||
}
|
||||
bundle := map[string]bool{"summary/ours": true, "entity/contested": true}
|
||||
if slug, collided := wikiSlugCollision(pages, []string{"summary/ours"}, bundle); !collided || slug != "entity/contested" {
|
||||
t.Fatalf("collision=%q %t", slug, collided)
|
||||
}
|
||||
if _, collided := wikiSlugCollision(pages, []string{"summary/ours", "entity/contested"}, bundle); collided {
|
||||
t.Fatal("in-scope slugs must not count as collisions")
|
||||
}
|
||||
// 包内的库级页（index/log）在降级路径上不写，因此目标库的同名页不算占用。
|
||||
library := []targetWikiRef{{ID: "p9", Slug: "index", SourceRefs: nil, ChunkRefs: nil}}
|
||||
if slug, collided := wikiSlugCollision(library, nil, map[string]bool{}); collided {
|
||||
t.Fatalf("library pages must not collide: %q", slug)
|
||||
}
|
||||
if slug, collided := wikiSlugCollision(library, nil, map[string]bool{"index": true}); !collided || slug != "index" {
|
||||
t.Fatalf("ref-bearing package page must collide: %q %t", slug, collided)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user