feat(release): 文档发布会带上知识库 wiki 快照并支持撤权剥离
发布通道原来只有三条：PostgreSQL 关系行、Elasticsearch 分片、OSS 对象。 WeKnora 的 wiki 图谱是 wiki_pages 一张纯 PG 表（in_links/out_links 存 slug）， 所以它不在任何通道里——撤权后生产 wiki 会一直留着已撤权文档的页面与引用。 本次新增第 5 条通道：库级 wiki 快照，挂在现有文档发布包里，检索与 wiki 同一个 release 单元（同一次执行、同一次校验、同一次回滚）： - 导出：ExportWiki 按 knowledge_base_id 全量读 wiki_pages/wiki_folders； 收敛门禁（库内无解析中/合成中文档）不满足直接失败并给中文原因； 未启用 wiki 且无 wiki 行的库不参与同步（生产 wiki 保持原样）。 - 撤权：item.Action == "delete" 时在导出端剥离该文档的 wiki 贡献 （source_refs/chunk_refs 去条目、无引用页面整页不带、摘死链、清 parent_slug、 无页面引用的空文件夹及其祖先一并去掉），使“源端仍在只停止发布”和 “源端删除”两种情况收敛到同一结果。 - 导入：库级 replace-by-kb（同一事务内 DELETE by kb 后灌入快照，文件夹先于页面）， 新增 ReplaceWiki 供撤权路径使用（文档行只 disable，wiki 整体替换）。 - 校验：VerifyWiki 用生产 WeKnora wiki API 比对页面 slug 集合，撤权时额外断言 生产 wiki 不再引用被撤权文档。 驱动与验收：新增 cmd/release-channel-e2e（复用发布 worker 真实代码路径， 支持 publish/withdraw/rollback），scripts/wiki_e2e_test5.sh、 scripts/wiki_e2e_newkb.sh 两套端到端脚本；证据见 docs/WIKI_CO_TRANSPORT_ACCEPTANCE.md。
This commit is contained in:
姚章浩
@@ -29,3 +29,5 @@
|
||||
.style-backup-*/
|
||||
.hermes/
|
||||
__pycache__/
|
||||
/dist/
|
||||
config/release.wiki-e2e.json
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
// Command release-channel-e2e drives the real v0.7.1 release channel end to end
|
||||
// against the endpoints in a release config file, so the retrieval lanes and the
|
||||
// knowledge base wiki snapshot can be exercised together:
|
||||
//
|
||||
// publish : Prepare → Activate → Verify (document + KB wiki into production)
|
||||
// withdraw : Prepare → Restore("absent") → VerifyRestored("absent")
|
||||
// (document disabled in retrieval AND its wiki contribution retracted)
|
||||
// rollback : Restore(revision) → VerifyRestored(revision)
|
||||
//
|
||||
// It reuses the same production code path as the release worker
|
||||
// (DocumentExporter → DocumentTarget → WeKnoraVerifier); only the orchestration
|
||||
// (platform queue rows) is skipped. Credentials are read from files and never
|
||||
// printed.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/governance"
|
||||
"knowledge-governance/internal/platform/weknora"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
releaseworkflow "knowledge-governance/internal/workflows/release"
|
||||
)
|
||||
|
||||
type objectCfg struct {
|
||||
Provider string `json:"provider"`
|
||||
StorageBackendID string `json:"storage_backend_id"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Region string `json:"region"`
|
||||
Bucket string `json:"bucket"`
|
||||
AccessKeyFile string `json:"access_key_file"`
|
||||
SecretKeyFile string `json:"secret_key_file"`
|
||||
}
|
||||
|
||||
type endpointCfg struct {
|
||||
DatabaseURLFile string `json:"database_url_file"`
|
||||
Elasticsearch struct {
|
||||
BaseURL string `json:"base_url"`
|
||||
Index string `json:"index"`
|
||||
TokenFile string `json:"token_file"`
|
||||
} `json:"elasticsearch"`
|
||||
Object objectCfg `json:"object"`
|
||||
WeKnora struct {
|
||||
BaseURL string `json:"base_url"`
|
||||
APIKeyFile string `json:"api_key_file"`
|
||||
} `json:"weknora"`
|
||||
}
|
||||
|
||||
type config struct {
|
||||
ArchiveRoot string `json:"archive_root"`
|
||||
Routes []struct {
|
||||
TenantID uint64 `json:"tenant_id"`
|
||||
KnowledgeBaseID string `json:"knowledge_base_id"`
|
||||
EmbeddingModel string `json:"embedding_model"`
|
||||
Source endpointCfg `json:"source"`
|
||||
Target endpointCfg `json:"target"`
|
||||
} `json:"routes"`
|
||||
}
|
||||
|
||||
func secret(path string) string {
|
||||
if path == "" {
|
||||
return ""
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
fmt.Println("FATAL read secret:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
func main() {
|
||||
configPath := flag.String("config", "", "release config json")
|
||||
knowledgeID := flag.String("knowledge", "", "knowledge id to release")
|
||||
mode := flag.String("mode", "publish", "publish | withdraw | rollback")
|
||||
revision := flag.String("revision", "wiki-e2e-v1", "release revision")
|
||||
previous := flag.String("previous", "", "previous production revision (rollback target)")
|
||||
archiveRoot := flag.String("archive", "", "override archive root")
|
||||
flag.Parse()
|
||||
if *configPath == "" || *knowledgeID == "" {
|
||||
fmt.Println("FATAL -config and -knowledge are required")
|
||||
os.Exit(2)
|
||||
}
|
||||
raw, err := os.ReadFile(*configPath)
|
||||
if err != nil {
|
||||
fmt.Println("FATAL read config:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var cfg config
|
||||
if err := json.Unmarshal(raw, &cfg); err != nil || len(cfg.Routes) != 1 {
|
||||
fmt.Println("FATAL parse config:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
route := cfg.Routes[0]
|
||||
if *archiveRoot != "" {
|
||||
cfg.ArchiveRoot = *archiveRoot
|
||||
}
|
||||
fmt.Printf("route kb=%s tenant=%d mode=%s revision=%s previous=%q\n", route.KnowledgeBaseID, route.TenantID, *mode, *revision, *previous)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second)
|
||||
defer cancel()
|
||||
|
||||
sourcePool, err := openPool(ctx, secret(route.Source.DatabaseURLFile))
|
||||
if err != nil {
|
||||
fmt.Println("STEP source PG connect: FAIL", err)
|
||||
return
|
||||
}
|
||||
defer sourcePool.Close()
|
||||
targetPool, err := openPool(ctx, secret(route.Target.DatabaseURLFile))
|
||||
if err != nil {
|
||||
fmt.Println("STEP target PG connect: FAIL", err)
|
||||
return
|
||||
}
|
||||
defer targetPool.Close()
|
||||
fmt.Println("STEP both PG connect: ok")
|
||||
|
||||
sourceSearch, err := v071.NewSearchExporter(v071.SearchExporterConfig{BaseURL: route.Source.Elasticsearch.BaseURL, Index: route.Source.Elasticsearch.Index, AuthorizationToken: secret(route.Source.Elasticsearch.TokenFile), PageSize: 500})
|
||||
if err != nil {
|
||||
fmt.Println("STEP source ES: FAIL", err)
|
||||
return
|
||||
}
|
||||
targetSearchConfig := v071.SearchExporterConfig{BaseURL: route.Target.Elasticsearch.BaseURL, Index: route.Target.Elasticsearch.Index, AuthorizationToken: secret(route.Target.Elasticsearch.TokenFile), PageSize: 500}
|
||||
targetSearchInspector, err := v071.NewSearchExporter(targetSearchConfig)
|
||||
if err != nil {
|
||||
fmt.Println("STEP target ES inspector: FAIL", err)
|
||||
return
|
||||
}
|
||||
targetSearch, err := v071.NewSearchImporter(targetSearchConfig)
|
||||
if err != nil {
|
||||
fmt.Println("STEP target ES importer: FAIL", err)
|
||||
return
|
||||
}
|
||||
sourceStore, err := newStore(route.Source.Object)
|
||||
if err != nil {
|
||||
fmt.Println("STEP source object store: FAIL", err)
|
||||
return
|
||||
}
|
||||
targetStore, err := newStore(route.Target.Object)
|
||||
if err != nil {
|
||||
fmt.Println("STEP target object store: FAIL", err)
|
||||
return
|
||||
}
|
||||
objectExporter, _ := v071.NewObjectExporter(sourceStore, v071.ObjectExporterLimits{})
|
||||
objectImporter, _ := v071.NewObjectImporter(targetStore, v071.TargetObjectConfig{StorageBackendID: route.Target.Object.StorageBackendID, Provider: route.Target.Object.Provider})
|
||||
postgresExporter, _ := v071.NewPostgresExporter(sourcePool)
|
||||
postgresImporter, err := v071.NewPostgresImporter(targetPool, v071.PostgresImporterConfig{TargetStorageBackendID: route.Target.Object.StorageBackendID, TargetProvider: route.Target.Object.Provider, TargetBucket: route.Target.Object.Bucket})
|
||||
if err != nil {
|
||||
fmt.Println("STEP target PG importer: FAIL", err)
|
||||
return
|
||||
}
|
||||
sourceContract, err := v071.NewTargetContractInspector(v071.TargetContractInspectorConfig{DB: sourcePool, Mapping: sourceSearch, KnowledgeBaseID: route.KnowledgeBaseID, TenantID: route.TenantID, IndexName: route.Source.Elasticsearch.Index})
|
||||
if err != nil {
|
||||
fmt.Println("STEP source contract: FAIL", err)
|
||||
return
|
||||
}
|
||||
targetContract, err := v071.NewTargetContractInspector(v071.TargetContractInspectorConfig{DB: targetPool, Mapping: targetSearchInspector, KnowledgeBaseID: route.KnowledgeBaseID, TenantID: route.TenantID, IndexName: route.Target.Elasticsearch.Index})
|
||||
if err != nil {
|
||||
fmt.Println("STEP target contract: FAIL", err)
|
||||
return
|
||||
}
|
||||
documentExporter, err := v071.NewDocumentExporter(v071.DocumentExporterConfig{Contract: sourceContract, Postgres: postgresExporter, Search: sourceSearch, Objects: objectExporter, TenantID: route.TenantID, EmbeddingModel: route.EmbeddingModel, Now: time.Now})
|
||||
if err != nil {
|
||||
fmt.Println("STEP document exporter: FAIL", err)
|
||||
return
|
||||
}
|
||||
archive, err := v071.NewBundleArchive(filepath.Join(cfg.ArchiveRoot, route.KnowledgeBaseID), 1024*1024*1024)
|
||||
if err != nil {
|
||||
fmt.Println("STEP archive: FAIL", err)
|
||||
return
|
||||
}
|
||||
weknoraClient, err := weknora.NewClient(weknora.ClientConfig{BaseURL: route.Target.WeKnora.BaseURL, APIKey: secret(route.Target.WeKnora.APIKeyFile), Timeout: 60 * time.Second, AllowDockerHTTP: true})
|
||||
if err != nil {
|
||||
fmt.Println("STEP target WeKnora client: FAIL", err)
|
||||
return
|
||||
}
|
||||
verifier, _ := v071.NewWeKnoraVerifier(weknoraClient)
|
||||
target, err := v071.NewDocumentTarget(v071.DocumentTargetConfig{Archive: archive, Contract: targetContract, Postgres: postgresImporter, Search: targetSearch, Objects: objectImporter, Verifier: verifier})
|
||||
if err != nil {
|
||||
fmt.Println("STEP document target: FAIL", err)
|
||||
return
|
||||
}
|
||||
|
||||
action := "upsert"
|
||||
if *mode == "withdraw" {
|
||||
action = "delete"
|
||||
}
|
||||
item := releaseworkflow.PlanItem{
|
||||
ReleaseExecutionID: "rx-wiki-e2e", PublicationID: "tp-wiki-e2e", Action: action,
|
||||
SourceItemVersionID: "sv-wiki-e2e", SourceRevision: "source-wiki-e2e",
|
||||
ParseRevision: "parser-wiki-e2e@index-wiki-e2e", ReleaseRevision: *revision,
|
||||
TestKnowledgeBaseID: route.KnowledgeBaseID, KnowledgeID: *knowledgeID,
|
||||
DecidedVisibility: governance.VisibilityInternal, PreviousProductionRevision: *previous,
|
||||
}
|
||||
printWiki("target wiki before", ctx, weknoraClient, route.KnowledgeBaseID)
|
||||
|
||||
switch *mode {
|
||||
case "publish":
|
||||
change, err := target.Prepare(ctx, item, mustExport(ctx, documentExporter, item))
|
||||
if err != nil {
|
||||
fmt.Println("STEP prepare: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Printf("STEP prepare: ok previous=%s production=%s\n", change.PreviousRevision, change.ProductionRevision)
|
||||
bundle, err := archive.Load(change.ProductionRevision)
|
||||
if err != nil {
|
||||
fmt.Println("STEP bundle reload: FAIL", err)
|
||||
return
|
||||
}
|
||||
printBundle(bundle)
|
||||
if err := target.Activate(ctx, item, change); err != nil {
|
||||
fmt.Println("STEP activate: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP activate: ok")
|
||||
if err := target.Verify(ctx, item, change); err != nil {
|
||||
fmt.Println("STEP verify: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP verify: ok (document + wiki)")
|
||||
case "withdraw":
|
||||
change, err := target.Prepare(ctx, item, mustExport(ctx, documentExporter, item))
|
||||
if err != nil {
|
||||
fmt.Println("STEP prepare: FAIL", err)
|
||||
return
|
||||
}
|
||||
bundle, err := archive.Load(change.ProductionRevision)
|
||||
if err != nil {
|
||||
fmt.Println("STEP bundle reload: FAIL", err)
|
||||
return
|
||||
}
|
||||
printBundle(bundle)
|
||||
revision, err := target.Restore(ctx, item, "absent")
|
||||
if err != nil {
|
||||
fmt.Println("STEP restore absent: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP restore absent: ok revision=", revision)
|
||||
if err := target.VerifyRestored(ctx, item, "absent"); err != nil {
|
||||
fmt.Println("STEP verify withdrawn: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP verify withdrawn: ok (retrieval disabled + wiki retracted)")
|
||||
case "rollback":
|
||||
if *previous == "" {
|
||||
fmt.Println("FATAL rollback mode needs -previous")
|
||||
os.Exit(2)
|
||||
}
|
||||
revision, err := target.Restore(ctx, item, *previous)
|
||||
if err != nil {
|
||||
fmt.Println("STEP restore: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP restore: ok revision=", revision)
|
||||
if err := target.VerifyRestored(ctx, item, *previous); err != nil {
|
||||
fmt.Println("STEP verify restored: FAIL", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("STEP verify restored: ok")
|
||||
default:
|
||||
fmt.Println("FATAL unknown mode:", *mode)
|
||||
os.Exit(2)
|
||||
}
|
||||
printWiki("target wiki after", ctx, weknoraClient, route.KnowledgeBaseID)
|
||||
}
|
||||
|
||||
func mustExport(ctx context.Context, exporter *v071.DocumentExporter, item releaseworkflow.PlanItem) releaseworkflow.ExportedBundle {
|
||||
value, err := exporter.Export(ctx, item)
|
||||
if err != nil {
|
||||
fmt.Println("STEP export: FAIL", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
func printBundle(bundle *v071.DocumentBundle) {
|
||||
parts := make([]string, 0, len(bundle.Postgres.Tables))
|
||||
for _, table := range bundle.Postgres.Tables {
|
||||
parts = append(parts, fmt.Sprintf("%s=%d", table.Name, table.RowCount))
|
||||
}
|
||||
fmt.Println("bundle tables:", strings.Join(parts, " "))
|
||||
fmt.Printf("bundle search=%d objects=%d\n", bundle.Manifest.Search.DocumentCount, bundle.Manifest.Assets.ObjectCount)
|
||||
}
|
||||
|
||||
func printWiki(label string, ctx context.Context, client *weknora.Client, knowledgeBaseID string) {
|
||||
pages, total, err := client.ListWikiPages(ctx, knowledgeBaseID, 1, 20)
|
||||
if err != nil {
|
||||
fmt.Printf("%s: FAIL %v\n", label, err)
|
||||
return
|
||||
}
|
||||
slugs := make([]string, 0, len(pages))
|
||||
for _, page := range pages {
|
||||
slugs = append(slugs, page.Slug)
|
||||
}
|
||||
fmt.Printf("%s: total=%d first=%s\n", label, total, strings.Join(slugs, ","))
|
||||
}
|
||||
|
||||
func newStore(object objectCfg) (*v071.S3ObjectStore, error) {
|
||||
parsed, err := url.Parse(object.Endpoint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
accessKey, secretKey := secret(object.AccessKeyFile), secret(object.SecretKeyFile)
|
||||
if accessKey == "" || secretKey == "" {
|
||||
return nil, errors.New("object credentials are required")
|
||||
}
|
||||
return v071.NewS3ObjectStore(v071.S3ObjectStoreConfig{Provider: object.Provider, Endpoint: object.Endpoint, Region: object.Region, AccessKeyID: accessKey, SecretAccessKey: secretKey, Bucket: object.Bucket, Secure: parsed.Scheme == "https"})
|
||||
}
|
||||
|
||||
func openPool(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) {
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
return pool, nil
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
# wiki 一同搬运与撤权（文档级发布单元内的库级 wiki 快照）
|
||||
|
||||
状态：实现中（本文件随实现同步更新）
|
||||
日期：2026-09-16
|
||||
范围：`internal/releasebundle`、`internal/releaseadapter/v071`、`internal/platform/weknora`
|
||||
|
||||
## 1. 目标（主人的原话）
|
||||
|
||||
> 测试的 weknora 解析后上传到生产，如果我在来源与目录停止了该知识，后面的生产发布环境能把这条文档去掉，包括检索和 wiki。
|
||||
|
||||
> OK 你帮我实现 wiki 一同搬运的需求。现在的知识库也得改成 wiki 知识库。然后用单文档进行测试，
|
||||
> 导入加删除，再 5 个文档测试。新建新的知识库 wiki。
|
||||
|
||||
拆成四件事：
|
||||
1. 发布时：文档的检索（PG+ES+OSS）与**该库的 wiki 页面**一起进生产。
|
||||
2. 撤权时：文档从生产检索里去掉的同时，**wiki 里这篇文档的贡献也去掉**。
|
||||
3. 现有知识库开启 wiki（改成 wiki 知识库）。
|
||||
4. 验证顺序：单文档（导入+删除）→ 5 文档 → 新建 wiki 知识库。
|
||||
|
||||
## 2. 现状与缺口
|
||||
|
||||
现状（已取证）：
|
||||
- 发布通道三条：PG 6 表（knowledges/chunks/knowledge_tags/knowledge_tag_relations/resources/resource_bindings）+ ES chunks/vectors + OSS 对象。
|
||||
- 撤权只做两件事：`UPDATE knowledges SET enable_status='disabled'` + ES delete-by-query。
|
||||
wiki 页面原样留在生产。
|
||||
- wiki 图谱不是 Neo4j：节点+边都在 `wiki_pages` 一张表里（`in_links`/`out_links` 存 slug 数组），
|
||||
文件夹在 `wiki_folders`。两者都是纯 PG，同 ID 可直搬，不需要重建、不需要 Neo4j。
|
||||
- `internal/releasebundle/manifest.go` 的白名单**已经**含 `wiki_pages`/`wiki_folders`，
|
||||
但导出/导入/校验/回滚全链路零实现（只有那两行白名单）。
|
||||
|
||||
缺口 = 第 5 条通道：**库级 wiki 快照**。
|
||||
|
||||
## 3. 设计（已按主人此前批准的方向实现）
|
||||
|
||||
### 3.1 快照载体：挂在文档发布包里，不新增发布单元
|
||||
|
||||
`wiki_pages` + `wiki_folders` 作为两张**库级（按 knowledge_base_id 全量）**表，追加进现有文档 bundle 的
|
||||
`PostgresSnapshot.Tables`。好处：
|
||||
- manifest 形状不变（`Tables` 白名单本来就允许这两张表），发布计划/编排/前端零改动。
|
||||
- 同一 release 单元：检索与 wiki 同一次执行、同一次校验、同一次回滚（决策④）。
|
||||
- 语义可判：manifest 里**有**这两张表 = 该包带 wiki 快照；**没有** = 老包/未启用 wiki（跳过 wiki 应用，不动生产 wiki）。
|
||||
|
||||
### 3.2 撤权时重新读源端 wiki（关键）
|
||||
|
||||
`internal/workflows/release/sidecar_publisher.go` 的 `Delete` 路径在 `Restore(item,"absent")` 之前
|
||||
**也会先 Export**，并把新包归档到 `item.ReleaseRevision`（`Prepare` → `Archive.Save`）。
|
||||
所以撤权时 `Restore("absent")` 加载到的是**撤权那一刻新鲜导出的包**，wiki 快照就是源端当前状态。
|
||||
不需要新增编排步骤。
|
||||
|
||||
### 3.3 撤权时对 wiki 做"该文档贡献剥离"（确定性，无 LLM）
|
||||
|
||||
导出时若 `item.Action == "delete"`：
|
||||
- `source_refs`（格式 `<knowledge_id>|<doc_title>`）里剥掉该 knowledge 的条目；
|
||||
- `chunk_refs` 里剥掉该 knowledge 的 chunk UUID；
|
||||
- 剥完 refs 后**没有任何来源**的页面 → 整页不带走（等同源端"专属页删除"）；
|
||||
- 索引页/日志页（无 refs）保留；
|
||||
- 指向被丢弃 slug 的 `in_links`/`out_links` 一并摘掉（避免生产死链），父页被丢弃时 `parent_slug` 清空。
|
||||
|
||||
为什么需要这层：源端"仅停用"不改 wiki（WeKnora 只在**删除**时跑 retract 清理链）。
|
||||
剥离层让"停止发布"和"源端删除"两种情况收敛到同一个结果：生产 wiki 不再引用该文档。
|
||||
源端真删除时，清理链已跑过，剥离是幂等空操作。
|
||||
|
||||
### 3.4 收敛门禁（决策①：等 + 拒出）
|
||||
|
||||
导出 wiki 前检查该库是否收敛：
|
||||
- 库内没有 `deleted_at IS NULL` 且 `pending_subtasks_count > 0` 的文档；
|
||||
- 库内没有 `parse_status IN ('pending','processing','finalizing')` 的文档。
|
||||
不收敛 → 导出失败并给出中文提示（不产出半成品 wiki 快照）。
|
||||
|
||||
### 3.5 生产端应用方式：库级 replace-by-kb（决策②③）
|
||||
|
||||
生产端在**同一个事务**里（沿用 `BEGIN → schema_migrations=74 断言 → advisory lock → COMMIT`）：
|
||||
`DELETE FROM wiki_pages/wiki_folders WHERE tenant_id=$1 AND knowledge_base_id=$2` → 灌入快照行。
|
||||
id/slug/in_links/out_links/folder_id 原样保留，行级 scope 校验（tenant_id + knowledge_base_id）。
|
||||
每次发布都推整库快照（幂等）。
|
||||
|
||||
### 3.6 未启用 wiki 的库不参与同步
|
||||
|
||||
源端库 `wiki_enabled=false` 且 wiki 行数为 0 → 包内**不带** wiki 表 → 生产 wiki 保持原样（不会被清空）。
|
||||
|
||||
## 4. 开发清单（代码）
|
||||
|
||||
| # | 文件 | 改动 |
|
||||
|---|---|---|
|
||||
| 1 | `internal/releaseadapter/v071/wiki_export.go`（新） | `PostgresExporter.ExportWiki`：收敛门禁 + 按 kb 读 wiki_pages/wiki_folders + 撤权剥离 |
|
||||
| 2 | `internal/releaseadapter/v071/document_export.go` | `validReleaseTable` 放行 wiki 两表；导出时追加 wiki 表 |
|
||||
| 3 | `internal/releaseadapter/v071/postgres_import.go` | `prepareSnapshot` 校验 wiki 行 scope；`Replace` 加库级 DELETE；新增 `ReplaceWiki`（仅 wiki） |
|
||||
| 4 | `internal/releaseadapter/v071/document_target.go` | `Restore("absent")` 追加 wiki 应用；`VerifyRestored` 追加 wiki 校验 |
|
||||
| 5 | `internal/releaseadapter/v071/verifier.go` | wiki 校验：生产 wiki API 的页数 + slug 集合与快照一致 |
|
||||
| 6 | `internal/platform/weknora/client.go`+`types.go` | 新增 `ListWikiPages`（`GET /api/v1/knowledgebase/{kb}/wiki/pages`，X-API-Key） |
|
||||
|
||||
## 5. 测试（每步都要运行证据）
|
||||
|
||||
| 步骤 | 内容 | 证据 |
|
||||
|---|---|---|
|
||||
| T0 | 单测 + vet + build | go test / vet / build 输出 |
|
||||
| T1 | 新建 wiki 知识库（`wiki_enabled=true`）+ 导入 1 篇，等 wiki 合成完成 | `wiki_pages` 行数 > 0 |
|
||||
| T2 | 单文档发布到生产（模拟生产 = 测试服 weknora-test-v071） | 生产 PG/ES/OSS/wiki 四通道行数 |
|
||||
| T3 | 源端删除该文档 → 等 wiki 收敛 → 撤权发布 | 生产该 knowledge `enable_status='disabled'`；ES chunks=0；wiki 该文档页/引用消失 |
|
||||
| T4 | 5 文档库：发布 → 删 1 篇并撤权 | 共贡页 refs 剥离、专属页消失、检索该篇=0 |
|
||||
| T5 | 新建 wiki 知识库再跑一遍单文档闭环 | 同 T2/T3 |
|
||||
|
||||
## 6. 验收标准
|
||||
|
||||
1. 发布后生产 wiki 页数/slug 集合 = 源端快照（含文件夹）。
|
||||
2. 撤权后生产检索无该文档（`enable_status='disabled'` + ES chunks=0）**且** wiki 无该文档贡献。
|
||||
3. 回滚到上一版 revision 时，wiki 一并回到上一版快照。
|
||||
4. 未启用 wiki 的库发布不改变生产 wiki。
|
||||
5. 源端 wiki 未收敛时导出**拒绝**并给出中文提示，不产生半成品。
|
||||
|
||||
## 7. 决策默认值（主人未反对即按此；要改告诉我）
|
||||
|
||||
| 点 | 采用值 |
|
||||
|---|---|
|
||||
| ① 撤权是否等 wiki 收敛 | 等：不收敛则**该次导出直接失败**（阻断这次发布/撤权）并给中文原因，不产出半成品快照 |
|
||||
| ② 覆盖语义 | 库级全量 replace（幂等、可回滚） |
|
||||
| ③ 推送粒度 | 每次发布/撤权都推整库快照 |
|
||||
| ④ 撤权顺序 | 检索与 wiki 同一 release 单元、同一事务链 |
|
||||
|
||||
## 8. 运行证据（2026-09-16，模拟生产 = 测试服 weknora-test-v071）
|
||||
|
||||
环境：源端 = 本机 WKNFull（KB `c12ff4d5-…`，wiki 打开）；目标端 = 测试服
|
||||
`weknora-test-v071`（RDS `weknora_test` + ES `weknora_test_chunks` + 生产桶
|
||||
`dobot-weknora-test`）。三条 ssh 隧道（25432/29200/28080）→ 本机 `prod-pg/prod-es/prod-weknora`。
|
||||
驱动：`cmd/release-channel-e2e`（复用发布 worker 的真实代码路径：
|
||||
DocumentExporter → DocumentTarget Prepare/Activate/Verify / Restore+VerifyRestored）。
|
||||
|
||||
### T1 单文档：导入 + 发布
|
||||
|
||||
- 源端：解析完成、wiki 子任务 0；`wiki_pages=7`。
|
||||
- 包内容：`knowledges=1 chunks=3 resources=1 resource_bindings=1 wiki_folders=5 wiki_pages=7`。
|
||||
- 结果：`STEP activate: ok` / `STEP verify: ok (document + wiki)`。
|
||||
- 生产 wiki：0 → 7 页（含 `index`、`summary/<knowledge>`、`entity/weknora` 等）。
|
||||
|
||||
### T2 单文档：撤权（源端文档仍在，只停止发布）
|
||||
|
||||
- 包内容：`wiki_pages=1`（只有 `index` 存活，专属页被剥掉）——撤权剥离在**导出端**生效。
|
||||
- 结果：`STEP restore absent: ok` / `STEP verify withdrawn: ok (retrieval disabled + wiki retracted)`。
|
||||
- 生产 wiki：7 → 1 页，且 `refs_still_citing=0`。
|
||||
- 生产检索：`enable=disabled`；ES `is_enabled=true` 分片 = 0（`is_enabled=false` = 2）。
|
||||
- 源端该文档 `parse_status=completed, deleted_at=null` → 证明是本通道剥离了 wiki，
|
||||
而不是源端清理链的副作用。
|
||||
|
||||
### T3 五文档：发布 + 撤其中一篇
|
||||
|
||||
- 生产：`enabled=5 / total=5`，`wiki_pages=23`，`wiki_folders=5`。
|
||||
- 共贡页面真实存在：`entity/weknora` 被 5 篇同时引用；`concept/publish-channel` 被 2 篇引用。
|
||||
- 撤权 doc `6d8d622d…` 后：`wiki_pages 23 → 20`，引用被撤权文档的页面 = 0，
|
||||
其余 4 篇仍 `enable=enabled`，ES 启用分片 = 2/2/2/3（合计 9，与该库剩余文档一致）。
|
||||
|
||||
### T4 新建 wiki 知识库：单文档导入 → 发布 → 撤权
|
||||
|
||||
脚本 `scripts/wiki_e2e_newkb.sh`：源/目标各克隆一个全新同 ID 知识库（`wiki-e2e-fresh`，开 wiki）。
|
||||
|
||||
- 导入后源端 wiki 收敛：`completed/0`，`wiki_pages=8`。
|
||||
- 发布包：`wiki_folders=7 wiki_pages=9`；`activate ok` / `verify ok`；生产 wiki 0 → 9 页。
|
||||
- 撤权后：该文档 `enable=disabled`；`wiki_pages=1 wiki_folders=0`
|
||||
（只剩 `index`，空文件夹一并清除）；`refs_citing_withdrawn=0`；ES 启用分片 = 0。
|
||||
|
||||
### 单元/集成验证
|
||||
|
||||
- `go build ./...` / `go vet ./...` 全绿；`go test ./...` 全绿。
|
||||
- 新增单测：`wiki_export_test.go`（收敛门禁、未启用库跳过、撤权剥离、文件夹剪枝、跨域拒绝）、
|
||||
`wiki_channel_test.go`（包内带 wiki、撤权传参、verifier 页集合与引用校验、撤权路径 ReplaceWiki+VerifyWiki）。
|
||||
|
||||
### 已知限制
|
||||
|
||||
1. 首次发布失败做补偿（Compensate）时不会还原 wiki：该路径下没有上一版 wiki 快照可比对，
|
||||
只把文档置为 disabled。回滚（rollback）不受影响——每个包都带整库快照。
|
||||
2. 页面里以对象存储定位符引用的图片仍走 `resources` 通道，wiki 正文中的图片 URL 不重写。
|
||||
3. 生产端 wiki 页面数超过 1000 时校验会失败（防失控预算），需要时调大上限。
|
||||
4. 撤权后生产桶里的对象不删（回滚要用），属既有行为。
|
||||
@@ -257,6 +257,25 @@ func (c *Client) ListChunks(ctx context.Context, knowledgeID string, page, pageS
|
||||
return response.Data, response.Total, nil
|
||||
}
|
||||
|
||||
// ListWikiPages lists the wiki pages of a knowledge base. Used by release
|
||||
// verification to confirm the production wiki matches the released snapshot.
|
||||
func (c *Client) ListWikiPages(ctx context.Context, knowledgeBaseID string, page, pageSize int) ([]WikiPageSummary, int64, error) {
|
||||
if !validPathSegment(knowledgeBaseID) || !validPage(page, pageSize) {
|
||||
return nil, 0, errors.New("invalid wiki page list request")
|
||||
}
|
||||
query := url.Values{}
|
||||
query.Set("page", fmt.Sprint(page))
|
||||
query.Set("page_size", fmt.Sprint(pageSize))
|
||||
var response wikiPageListResponse
|
||||
if err := c.doJSONQuery(ctx, http.MethodGet, "/api/v1/knowledgebase/"+knowledgeBaseID+"/wiki/pages", query, nil, &response); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if response.Total < 0 {
|
||||
return nil, 0, errors.New("WeKnora returned an unsuccessful wiki page list")
|
||||
}
|
||||
return response.Pages, response.Total, nil
|
||||
}
|
||||
|
||||
func (c *Client) SearchKnowledge(ctx context.Context, request SearchRequest) ([]SearchResult, error) {
|
||||
if strings.TrimSpace(request.Query) == "" || len(request.KnowledgeIDs) == 0 {
|
||||
return nil, errors.New("search query and knowledge IDs are required")
|
||||
|
||||
@@ -88,3 +88,27 @@ type searchResponse struct {
|
||||
Success bool `json:"success"`
|
||||
Data []SearchResult `json:"data"`
|
||||
}
|
||||
|
||||
// WikiPageSummary is the wiki page projection the release verifier needs:
|
||||
// the page identity (slug) plus the source document references that back it.
|
||||
// WeKnora's wiki page list endpoint returns the full page objects; only these
|
||||
// fields are consulted during verification.
|
||||
type WikiPageSummary struct {
|
||||
Slug string `json:"slug"`
|
||||
Title string `json:"title"`
|
||||
PageType string `json:"page_type"`
|
||||
Status string `json:"status"`
|
||||
SourceRefs []string `json:"source_refs"`
|
||||
ChunkRefs []string `json:"chunk_refs"`
|
||||
Version int `json:"version"`
|
||||
}
|
||||
|
||||
// wikiPageListResponse mirrors types.WikiPageListResponse in WeKnora v0.7.1.
|
||||
// The wiki endpoints return the payload unwrapped (no {success,data} envelope).
|
||||
type wikiPageListResponse struct {
|
||||
Pages []WikiPageSummary `json:"pages"`
|
||||
Total int64 `json:"total"`
|
||||
Page int `json:"page"`
|
||||
PageSize int `json:"page_size"`
|
||||
TotalPages int `json:"total_pages"`
|
||||
}
|
||||
|
||||
@@ -94,6 +94,22 @@ func (exporter *DocumentExporter) Export(ctx context.Context, item releaseworkfl
|
||||
if !validLowerSHA256(objectSnapshot.SHA256) {
|
||||
return nil, errors.New("object document snapshot is incomplete")
|
||||
}
|
||||
// 库级 wiki 快照与本文档同一个发布单元：检索（PG/ES/OSS）和 wiki 一起进生产。
|
||||
// 撤权（delete）时剥掉该文档在 wiki 里的贡献，避免生产 wiki 继续引用已撤权的文档。
|
||||
if wikiExporter, ok := exporter.config.Postgres.(wikiSnapshotExporter); ok {
|
||||
removingKnowledgeID := ""
|
||||
if item.Action == "delete" {
|
||||
removingKnowledgeID = item.KnowledgeID
|
||||
}
|
||||
wikiTables, err := wikiExporter.ExportWiki(ctx, PostgresIdentity{
|
||||
TenantID: exporter.config.TenantID, KnowledgeBaseID: item.TestKnowledgeBaseID, KnowledgeID: item.KnowledgeID,
|
||||
}, removingKnowledgeID)
|
||||
if err != nil {
|
||||
// 例如源端 wiki 未收敛，直接透出面向运营的中文原因。
|
||||
return nil, err
|
||||
}
|
||||
postgresSnapshot.Tables = append(postgresSnapshot.Tables, wikiTables...)
|
||||
}
|
||||
tables := make([]releasebundle.TableDigest, 0, len(postgresSnapshot.Tables))
|
||||
for _, table := range postgresSnapshot.Tables {
|
||||
if !validReleaseTable(table.Name) || table.RowCount != len(table.Rows) || !validLowerSHA256(table.SHA256) {
|
||||
@@ -139,7 +155,8 @@ func validDocumentPlan(item releaseworkflow.PlanItem) bool {
|
||||
|
||||
func validReleaseTable(name string) bool {
|
||||
switch name {
|
||||
case "knowledges", "chunks", "knowledge_tags", "knowledge_tag_relations", "resources", "resource_bindings":
|
||||
case "knowledges", "chunks", "knowledge_tags", "knowledge_tag_relations", "resources", "resource_bindings",
|
||||
"wiki_pages", "wiki_folders":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
|
||||
@@ -20,6 +20,9 @@ type targetContractInspector interface {
|
||||
type postgresDocumentTarget interface {
|
||||
Replace(ctx context.Context, identity PostgresIdentity, snapshot PostgresSnapshot) error
|
||||
Disable(ctx context.Context, identity PostgresIdentity) error
|
||||
// ReplaceWiki 只应用库级 wiki 快照：撤权/回滚路径上文档行不能重灌，
|
||||
// 但 wiki 必须整体替换成目标版本的快照。
|
||||
ReplaceWiki(ctx context.Context, identity PostgresIdentity, snapshot PostgresSnapshot) error
|
||||
}
|
||||
|
||||
type searchDocumentTarget interface {
|
||||
@@ -35,6 +38,8 @@ type objectDocumentTarget interface {
|
||||
type ProductionVerifier interface {
|
||||
Verify(ctx context.Context, item releaseworkflow.PlanItem, bundle *DocumentBundle) error
|
||||
VerifyDisabled(ctx context.Context, item releaseworkflow.PlanItem) error
|
||||
// VerifyWiki 校验生产端 wiki 与包内快照一致（老包/未启用 wiki 时跳过）。
|
||||
VerifyWiki(ctx context.Context, item releaseworkflow.PlanItem, bundle *DocumentBundle) error
|
||||
}
|
||||
|
||||
type DocumentTargetConfig struct {
|
||||
@@ -117,6 +122,9 @@ func (target *DocumentTarget) Verify(ctx context.Context, item releaseworkflow.P
|
||||
if err := target.config.Verifier.Verify(ctx, item, bundle); err != nil {
|
||||
return errors.New("verify v0.7.1 production release")
|
||||
}
|
||||
if err := target.config.Verifier.VerifyWiki(ctx, item, bundle); err != nil {
|
||||
return errors.New("verify v0.7.1 production wiki release")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -154,6 +162,13 @@ func (target *DocumentTarget) Restore(ctx context.Context, item releaseworkflow.
|
||||
if err := target.config.Search.Disable(ctx, searchIdentity); err != nil {
|
||||
return "", errors.New("disable current Elasticsearch release")
|
||||
}
|
||||
// 文档行只做 disable，但 wiki 要整体替换成撤权后的快照，
|
||||
// 否则生产 wiki 会继续留着这条文档的页面与引用。
|
||||
if hasWikiSnapshot(current.Postgres) {
|
||||
if err := target.config.Postgres.ReplaceWiki(ctx, postgresIdentity, current.Postgres); err != nil {
|
||||
return "", errors.New("apply withdrawn v0.7.1 wiki snapshot")
|
||||
}
|
||||
}
|
||||
return "absent", nil
|
||||
}
|
||||
if err := target.restoreBundle(ctx, revision); err != nil {
|
||||
@@ -167,6 +182,13 @@ func (target *DocumentTarget) VerifyRestored(ctx context.Context, item releasewo
|
||||
if err := target.config.Verifier.VerifyDisabled(ctx, item); err != nil {
|
||||
return errors.New("verify disabled production release")
|
||||
}
|
||||
current, err := target.config.Archive.Load(item.ReleaseRevision)
|
||||
if err != nil {
|
||||
return errors.New("load withdrawn v0.7.1 release for wiki verification")
|
||||
}
|
||||
if err := target.config.Verifier.VerifyWiki(ctx, item, current); err != nil {
|
||||
return errors.New("verify withdrawn v0.7.1 wiki release")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
bundle, err := target.config.Archive.Load(revision)
|
||||
@@ -176,6 +198,9 @@ func (target *DocumentTarget) VerifyRestored(ctx context.Context, item releasewo
|
||||
if err := target.config.Verifier.Verify(ctx, item, bundle); err != nil {
|
||||
return errors.New("verify restored v0.7.1 release")
|
||||
}
|
||||
if err := target.config.Verifier.VerifyWiki(ctx, item, bundle); err != nil {
|
||||
return errors.New("verify restored v0.7.1 wiki release")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -48,7 +48,7 @@ func TestDocumentTargetCoordinatesPrepareActivateVerifyAndRollback(t *testing.T)
|
||||
if err := target.VerifyRestored(context.Background(), item, "prod-v0"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{"object-stage:review-v1", "object-stage:review-v1", "object-promote:review-v1", "postgres-replace", "search-replace", "verify:review-v1", "object-stage:prod-v0", "object-promote:prod-v0", "postgres-replace", "search-replace", "object-stage:prod-v0", "object-promote:prod-v0", "postgres-replace", "search-replace", "verify:prod-v0"}
|
||||
want := []string{"object-stage:review-v1", "object-stage:review-v1", "object-promote:review-v1", "postgres-replace", "search-replace", "verify:review-v1", "verify-wiki:review-v1", "object-stage:prod-v0", "object-promote:prod-v0", "postgres-replace", "search-replace", "object-stage:prod-v0", "object-promote:prod-v0", "postgres-replace", "search-replace", "verify:prod-v0", "verify-wiki:prod-v0"}
|
||||
if !reflect.DeepEqual(calls, want) {
|
||||
t.Fatalf("calls=%v\nwant=%v", calls, want)
|
||||
}
|
||||
@@ -99,6 +99,10 @@ func (s *stubPostgresTarget) Disable(context.Context, PostgresIdentity) error {
|
||||
*s.calls = append(*s.calls, "postgres-disable")
|
||||
return s.err
|
||||
}
|
||||
func (s *stubPostgresTarget) ReplaceWiki(_ context.Context, _ PostgresIdentity, _ PostgresSnapshot) error {
|
||||
*s.calls = append(*s.calls, "postgres-replace-wiki")
|
||||
return s.err
|
||||
}
|
||||
|
||||
type stubSearchTarget struct {
|
||||
calls *[]string
|
||||
@@ -141,6 +145,10 @@ func (s *stubProductionVerifier) VerifyDisabled(context.Context, releaseworkflow
|
||||
*s.calls = append(*s.calls, "verify-disabled")
|
||||
return s.err
|
||||
}
|
||||
func (s *stubProductionVerifier) VerifyWiki(_ context.Context, _ releaseworkflow.PlanItem, bundle *DocumentBundle) error {
|
||||
*s.calls = append(*s.calls, "verify-wiki:"+bundle.Manifest.ReleaseRevision)
|
||||
return s.err
|
||||
}
|
||||
|
||||
func cloneBundleWithRevision(t *testing.T, bundle *DocumentBundle, revision string) *DocumentBundle {
|
||||
t.Helper()
|
||||
|
||||
@@ -32,6 +32,12 @@ type PostgresImporter struct {
|
||||
config PostgresImporterConfig
|
||||
}
|
||||
|
||||
// clearStatement 是导入前置清理语句（文档级或库级 wiki）。
|
||||
type clearStatement struct {
|
||||
sql string
|
||||
args []any
|
||||
}
|
||||
|
||||
func NewPostgresImporter(db postgresTransactionalDB, options ...PostgresImporterConfig) (*PostgresImporter, error) {
|
||||
if db == nil || len(options) > 1 {
|
||||
return nil, errors.New("invalid PostgreSQL importer configuration")
|
||||
@@ -71,6 +77,13 @@ func (importer *PostgresImporter) Replace(ctx context.Context, identity Postgres
|
||||
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, identity.KnowledgeID); err != nil {
|
||||
return errors.New("lock target knowledge")
|
||||
}
|
||||
_, hasWiki := wikiSnapshotRows(prepared)
|
||||
if hasWiki {
|
||||
// 库级 wiki 替换加知识库维度的锁（命名空间 1），与文档维度的锁区分开。
|
||||
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")
|
||||
}
|
||||
}
|
||||
var crossScope int
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT COUNT(*)::INTEGER
|
||||
@@ -101,19 +114,29 @@ WHERE (id = $1 OR handle = $2)
|
||||
return errors.New("target knowledge tag identity conflict")
|
||||
}
|
||||
}
|
||||
for _, statement := range []struct {
|
||||
sql string
|
||||
args []any
|
||||
}{
|
||||
clearStatements := []clearStatement{
|
||||
{`DELETE FROM resource_bindings WHERE tenant_id = $1 AND ((owner_type = 'knowledge' AND owner_id = $2) OR (owner_type = 'chunk' AND owner_id = ANY($3::text[])))`, []any{identity.TenantID, identity.KnowledgeID, chunkIDs}},
|
||||
{`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")
|
||||
}
|
||||
}
|
||||
for _, tableName := range []string{"knowledges", "resources", "chunks", "knowledge_tags", "knowledge_tag_relations", "resource_bindings"} {
|
||||
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
|
||||
@@ -128,6 +151,65 @@ WHERE (id = $1 OR handle = $2)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReplaceWiki 只应用库级 wiki 快照（撤权 / 回滚到旧版时使用）。
|
||||
// 撤权路径上文档行只做 disable（不能把文档行重新灌回去），而 wiki 必须整体
|
||||
// 替换成撤权后的快照，否则生产 wiki 会一直留着已撤权文档的页面与引用。
|
||||
func (importer *PostgresImporter) ReplaceWiki(ctx context.Context, identity PostgresIdentity, snapshot PostgresSnapshot) error {
|
||||
if identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) {
|
||||
return errors.New("invalid PostgreSQL wiki import identity")
|
||||
}
|
||||
tables, hasWiki := wikiSnapshotRows(snapshot)
|
||||
if !hasWiki {
|
||||
// 老包 / 未启用 wiki 的库：不动生产 wiki。
|
||||
return nil
|
||||
}
|
||||
for _, table := range tables {
|
||||
if !validReleaseTable(table.Name) || table.RowCount != len(table.Rows) || newTableSnapshot(table.Name, table.Rows).SHA256 != table.SHA256 {
|
||||
return errors.New("PostgreSQL wiki import snapshot is invalid")
|
||||
}
|
||||
}
|
||||
for _, row := range snapshot.Table("wiki_pages").Rows {
|
||||
if !matchesWikiPageRow(row, identity) {
|
||||
return errors.New("PostgreSQL wiki page scope mismatch")
|
||||
}
|
||||
}
|
||||
for _, row := range snapshot.Table("wiki_folders").Rows {
|
||||
if !matchesWikiFolderRow(row, identity) {
|
||||
return errors.New("PostgreSQL wiki folder scope mismatch")
|
||||
}
|
||||
}
|
||||
tx, err := importer.db.Begin(ctx)
|
||||
if err != nil {
|
||||
return errors.New("begin PostgreSQL wiki import")
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var version int
|
||||
var dirty bool
|
||||
if err := tx.QueryRow(ctx, `SELECT version, dirty FROM schema_migrations LIMIT 1`).Scan(&version, &dirty); err != nil || version != 74 || dirty {
|
||||
return errors.New("target WeKnora schema contract mismatch")
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return errors.New("commit PostgreSQL wiki import")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (importer *PostgresImporter) Disable(ctx context.Context, identity PostgresIdentity) error {
|
||||
if identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) || !validSearchIdentifier(identity.KnowledgeID) {
|
||||
return errors.New("invalid PostgreSQL disable identity")
|
||||
@@ -263,6 +345,17 @@ func (importer *PostgresImporter) prepareSnapshot(identity PostgresIdentity, sna
|
||||
return PostgresSnapshot{}, nil, nil, errors.New("PostgreSQL import resource binding scope mismatch")
|
||||
}
|
||||
}
|
||||
// wiki 是库级快照：行必须落在同一个 tenant + knowledge base 里。
|
||||
for _, row := range prepared.Table("wiki_pages").Rows {
|
||||
if !matchesWikiPageRow(row, identity) {
|
||||
return PostgresSnapshot{}, nil, nil, errors.New("PostgreSQL import wiki page scope mismatch")
|
||||
}
|
||||
}
|
||||
for _, row := range prepared.Table("wiki_folders").Rows {
|
||||
if !matchesWikiFolderRow(row, identity) {
|
||||
return PostgresSnapshot{}, nil, nil, errors.New("PostgreSQL import wiki folder scope mismatch")
|
||||
}
|
||||
}
|
||||
return prepared, chunkIDs, resourcePairs, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ type WeKnoraVerificationClient interface {
|
||||
GetKnowledge(ctx context.Context, knowledgeID string) (weknora.Knowledge, error)
|
||||
ListChunks(ctx context.Context, knowledgeID string, page, pageSize int, chunkTypes []string) ([]weknora.Chunk, int64, error)
|
||||
SearchKnowledge(ctx context.Context, request weknora.SearchRequest) ([]weknora.SearchResult, error)
|
||||
ListWikiPages(ctx context.Context, knowledgeBaseID string, page, pageSize int) ([]weknora.WikiPageSummary, int64, error)
|
||||
}
|
||||
|
||||
type WeKnoraVerifier struct {
|
||||
@@ -138,4 +139,69 @@ func boundedProbeText(value string) string {
|
||||
return string(runes[:128])
|
||||
}
|
||||
|
||||
// VerifyWiki 校验生产端 wiki 与本次快照一致：
|
||||
// - 页面 slug 集合必须与快照完全相同（多一个少一个都算失败，这样“撤权后
|
||||
// 生产还留着已撤权文档的页面”会被抓住）；
|
||||
// - 撤权（delete）时生产 wiki 任何页面都不得再引用被撤权的文档。
|
||||
//
|
||||
// 老包 / 未启用 wiki 的库不携带 wiki 快照，直接跳过（生产 wiki 保持原样）。
|
||||
func (verifier *WeKnoraVerifier) VerifyWiki(ctx context.Context, item releaseworkflow.PlanItem, bundle *DocumentBundle) error {
|
||||
if bundle == nil || validateDocumentBundle(bundle) != nil {
|
||||
return errors.New("invalid WeKnora wiki verification bundle")
|
||||
}
|
||||
if !hasWikiSnapshot(bundle.Postgres) {
|
||||
return nil
|
||||
}
|
||||
expected := map[string]bool{}
|
||||
for _, raw := range bundle.Postgres.Table("wiki_pages").Rows {
|
||||
var row struct {
|
||||
Slug string `json:"slug"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || strings.TrimSpace(row.Slug) == "" {
|
||||
return errors.New("invalid WeKnora wiki verification bundle")
|
||||
}
|
||||
expected[row.Slug] = true
|
||||
}
|
||||
const pageSize = 20
|
||||
const maxPageBudget = 1000
|
||||
seen := map[string]bool{}
|
||||
scanned := 0
|
||||
for page := 1; ; page++ {
|
||||
pages, total, err := verifier.client.ListWikiPages(ctx, item.TestKnowledgeBaseID, page, pageSize)
|
||||
if err != nil {
|
||||
return errors.New("production WeKnora wiki verification failed")
|
||||
}
|
||||
for _, wikiPage := range pages {
|
||||
if strings.TrimSpace(wikiPage.Slug) == "" {
|
||||
continue
|
||||
}
|
||||
seen[wikiPage.Slug] = true
|
||||
if item.Action != "delete" {
|
||||
continue
|
||||
}
|
||||
for _, reference := range wikiPage.SourceRefs {
|
||||
if referencesKnowledge(reference, item.KnowledgeID) {
|
||||
return errors.New("production WeKnora wiki still references the withdrawn document")
|
||||
}
|
||||
}
|
||||
}
|
||||
scanned += len(pages)
|
||||
if len(pages) == 0 || int64(scanned) >= total {
|
||||
break
|
||||
}
|
||||
if page >= maxPageBudget {
|
||||
return errors.New("production WeKnora wiki verification exceeded its page budget")
|
||||
}
|
||||
}
|
||||
if len(seen) != len(expected) {
|
||||
return errors.New("production WeKnora wiki page set mismatch")
|
||||
}
|
||||
for slug := range expected {
|
||||
if !seen[slug] {
|
||||
return errors.New("production WeKnora wiki page set mismatch")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
var _ ProductionVerifier = (*WeKnoraVerifier)(nil)
|
||||
|
||||
@@ -47,6 +47,27 @@ type verificationClient struct {
|
||||
search []weknora.SearchResult
|
||||
searchErr error
|
||||
searchRequest weknora.SearchRequest
|
||||
wikiPages []weknora.WikiPageSummary
|
||||
wikiTotal int64
|
||||
wikiErr error
|
||||
}
|
||||
|
||||
func (c *verificationClient) ListWikiPages(_ context.Context, _ string, page, pageSize int) ([]weknora.WikiPageSummary, int64, error) {
|
||||
if c.wikiErr != nil {
|
||||
return nil, 0, c.wikiErr
|
||||
}
|
||||
if pageSize <= 0 {
|
||||
pageSize = 20
|
||||
}
|
||||
start := (page - 1) * pageSize
|
||||
if start >= len(c.wikiPages) {
|
||||
return nil, c.wikiTotal, nil
|
||||
}
|
||||
end := start + pageSize
|
||||
if end > len(c.wikiPages) {
|
||||
end = len(c.wikiPages)
|
||||
}
|
||||
return c.wikiPages[start:end], c.wikiTotal, nil
|
||||
}
|
||||
|
||||
func (c *verificationClient) GetKnowledge(context.Context, string) (weknora.Knowledge, error) {
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"knowledge-governance/internal/platform/weknora"
|
||||
releaseworkflow "knowledge-governance/internal/workflows/release"
|
||||
)
|
||||
|
||||
// stubPGWikiExporter 在文档导出器之上模拟“库级 wiki 快照”能力。
|
||||
type stubPGWikiExporter struct {
|
||||
stubPGExporter
|
||||
wikiTables []TableSnapshot
|
||||
removing string
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubPGWikiExporter) ExportWiki(_ context.Context, _ PostgresIdentity, removingKnowledgeID string) ([]TableSnapshot, error) {
|
||||
s.removing = removingKnowledgeID
|
||||
return s.wikiTables, s.err
|
||||
}
|
||||
|
||||
func wikiTablesFor(t *testing.T, slugs ...string) []TableSnapshot {
|
||||
t.Helper()
|
||||
rows := make([]json.RawMessage, 0, len(slugs))
|
||||
for _, slug := range slugs {
|
||||
rows = append(rows, mustCanonicalRows(t, `{"id":"page-`+slug+`","tenant_id":10000,"knowledge_base_id":"kb-safe","slug":"`+slug+`","title":"`+slug+`","source_refs":["knowledge-safe|a.md"],"in_links":[],"out_links":[]}`)[0])
|
||||
}
|
||||
return []TableSnapshot{newTableSnapshot("wiki_folders", nil), newTableSnapshot("wiki_pages", rows)}
|
||||
}
|
||||
|
||||
// wikiCarryingBundle 通过真实导出器产出一个带 wiki 快照的合法 bundle（manifest 与表摘要一致）。
|
||||
func wikiCarryingBundle(t *testing.T, wikiExporter *stubPGWikiExporter, action string) (*DocumentBundle, releaseworkflow.PlanItem) {
|
||||
t.Helper()
|
||||
pg := PostgresSnapshot{Tables: []TableSnapshot{
|
||||
newTableSnapshot("knowledges", mustCanonicalRows(t, `{"id":"knowledge-safe","title":"synthetic release marker"}`)),
|
||||
newTableSnapshot("chunks", mustCanonicalRows(t, `{"id":"chunk-safe"}`)),
|
||||
}}
|
||||
search := SearchSnapshot{Documents: []SearchDocument{{
|
||||
ID: "es-safe", Source: mustCanonicalRows(t, `{"tenant_id":10000,"knowledge_base_id":"kb-safe","knowledge_id":"knowledge-safe","content":"synthetic release marker"}`)[0],
|
||||
}}}
|
||||
search.DocumentsSHA256, _ = searchDocumentsDigest(search.Documents)
|
||||
search.MappingSHA256 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
objects := ObjectSnapshot{SHA256: objectSnapshotDigest(nil)}
|
||||
wikiExporter.stubPGExporter = stubPGExporter{snapshot: pg}
|
||||
exporter, err := NewDocumentExporter(DocumentExporterConfig{
|
||||
Contract: &stubSourceContract{contract: sourceContractFixture(search.MappingSHA256)},
|
||||
Postgres: wikiExporter, Search: &stubSearchExporter{snapshot: search},
|
||||
Objects: &stubObjectExporter{snapshot: objects}, TenantID: 10000, EmbeddingModel: "bge-m3", Now: fixedArchiveTime,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
item := releasePlanForArchive()
|
||||
item.Action = action
|
||||
value, err := exporter.Export(context.Background(), item)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return value.(*DocumentBundle), item
|
||||
}
|
||||
|
||||
func TestDocumentExporterCarriesWikiSnapshotIntoBundle(t *testing.T) {
|
||||
wikiExporter := &stubPGWikiExporter{wikiTables: wikiTablesFor(t, "entity/alpha", "index")}
|
||||
bundle, _ := wikiCarryingBundle(t, wikiExporter, "upsert")
|
||||
if wikiExporter.removing != "" {
|
||||
t.Fatalf("publish must not retract: %q", wikiExporter.removing)
|
||||
}
|
||||
if bundle.Manifest.Tables == nil {
|
||||
t.Fatal("missing manifest tables")
|
||||
}
|
||||
if got := bundle.Postgres.Table("wiki_pages").RowCount; got != 2 {
|
||||
t.Fatalf("wiki_pages rows=%d", got)
|
||||
}
|
||||
if !hasWikiSnapshot(bundle.Postgres) {
|
||||
t.Fatal("bundle must carry a wiki snapshot")
|
||||
}
|
||||
sealed := false
|
||||
for _, table := range bundle.Manifest.Tables {
|
||||
sealed = sealed || table.Name == "wiki_pages"
|
||||
}
|
||||
if !sealed {
|
||||
t.Fatalf("manifest=%#v", bundle.Manifest.Tables)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentExporterRetractsWithdrawnDocumentFromWiki(t *testing.T) {
|
||||
wikiExporter := &stubPGWikiExporter{wikiTables: wikiTablesFor(t, "entity/alpha")}
|
||||
if _, _ = wikiCarryingBundle(t, wikiExporter, "delete"); wikiExporter.removing != "knowledge-safe" {
|
||||
t.Fatalf("removing=%q", wikiExporter.removing)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentExporterSkipsWikiWhenKnowledgeBaseHasNone(t *testing.T) {
|
||||
wikiExporter := &stubPGWikiExporter{}
|
||||
bundle, _ := wikiCarryingBundle(t, wikiExporter, "upsert")
|
||||
if hasWikiSnapshot(bundle.Postgres) {
|
||||
t.Fatalf("unexpected wiki snapshot: %#v", bundle.Postgres.Tables)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeKnoraVerifierChecksWikiPageSet(t *testing.T) {
|
||||
bundle, item := wikiCarryingBundle(t, &stubPGWikiExporter{wikiTables: wikiTablesFor(t, "entity/alpha", "index")}, "upsert")
|
||||
|
||||
client := &verificationClient{wikiPages: []weknora.WikiPageSummary{{Slug: "entity/alpha"}, {Slug: "index"}}, wikiTotal: 2}
|
||||
verifier, _ := NewWeKnoraVerifier(client)
|
||||
if err := verifier.VerifyWiki(context.Background(), item, bundle); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
client.wikiPages = append(client.wikiPages, weknora.WikiPageSummary{Slug: "entity/stale"})
|
||||
client.wikiTotal = 3
|
||||
if err := verifier.VerifyWiki(context.Background(), item, bundle); err == nil {
|
||||
t.Fatal("stale production wiki page accepted")
|
||||
}
|
||||
|
||||
client.wikiPages = []weknora.WikiPageSummary{{Slug: "entity/alpha"}}
|
||||
client.wikiTotal = 1
|
||||
if err := verifier.VerifyWiki(context.Background(), item, bundle); err == nil {
|
||||
t.Fatal("missing production wiki page accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeKnoraVerifierRejectsProductionWikiReferringToWithdrawnDocument(t *testing.T) {
|
||||
bundle, item := wikiCarryingBundle(t, &stubPGWikiExporter{wikiTables: wikiTablesFor(t, "entity/shared")}, "delete")
|
||||
item.KnowledgeID = "knowledge-safe"
|
||||
client := &verificationClient{
|
||||
wikiPages: []weknora.WikiPageSummary{{Slug: "entity/shared", SourceRefs: []string{"knowledge-other|b.md"}}},
|
||||
wikiTotal: 1,
|
||||
}
|
||||
verifier, _ := NewWeKnoraVerifier(client)
|
||||
if err := verifier.VerifyWiki(context.Background(), item, bundle); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
client.wikiPages[0].SourceRefs = []string{"knowledge-other|b.md", "knowledge-safe|a.md"}
|
||||
if err := verifier.VerifyWiki(context.Background(), item, bundle); err == nil {
|
||||
t.Fatal("production wiki still citing the withdrawn document accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeKnoraVerifierSkipsWikiForBundlesWithoutSnapshot(t *testing.T) {
|
||||
bundle := testDocumentBundle(t)
|
||||
client := &verificationClient{wikiErr: errUnexpectedWikiCall()}
|
||||
verifier, _ := NewWeKnoraVerifier(client)
|
||||
if err := verifier.VerifyWiki(context.Background(), releasePlanForArchive(), bundle); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func errUnexpectedWikiCall() error { return os.ErrInvalid }
|
||||
|
||||
func TestDocumentTargetAppliesWikiSnapshotWhenWithdrawing(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
_ = os.Chmod(root, 0o700)
|
||||
archive, _ := NewBundleArchive(root, 16*1024*1024)
|
||||
bundle, item := wikiCarryingBundle(t, &stubPGWikiExporter{wikiTables: wikiTablesFor(t, "entity/shared")}, "delete")
|
||||
if err := archive.Save(bundle); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
calls := []string{}
|
||||
target, err := NewDocumentTarget(DocumentTargetConfig{
|
||||
Archive: archive, Contract: &stubContractInspector{contract: targetContractForBundle(bundle)},
|
||||
Postgres: &stubPostgresTarget{calls: &calls}, Search: &stubSearchTarget{calls: &calls},
|
||||
Objects: &stubObjectTarget{calls: &calls}, Verifier: &stubProductionVerifier{calls: &calls},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
revision, err := target.Restore(context.Background(), item, "absent")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if revision != "absent" {
|
||||
t.Fatalf("revision=%q", revision)
|
||||
}
|
||||
if err := target.VerifyRestored(context.Background(), item, "absent"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{"postgres-disable", "search-disable", "postgres-replace-wiki", "verify-disabled", "verify-wiki:review-v1"}
|
||||
if !reflect.DeepEqual(calls, want) {
|
||||
t.Fatalf("calls=%v\nwant=%v", calls, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,312 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 库级 wiki 快照（第 5 条发布通道）。
|
||||
//
|
||||
// WeKnora 的 wiki 图谱不是 Neo4j 图谱：节点与边都在 wiki_pages 一张表里
|
||||
// （in_links/out_links 存 slug 数组，库内相对引用），文件夹在 wiki_folders。
|
||||
// 两者都是纯 PostgreSQL 数据，同一个 knowledge_base_id 可以直接搬运，
|
||||
// 不需要重建，也不依赖 Neo4j / Elasticsearch。
|
||||
//
|
||||
// 快照粒度是“整库”，不是“单文档”：撤一篇文档会改动其它共贡页面的正文
|
||||
// （retract 重写、摘死链、重建索引页），所以 delta 的最小单位只能是整库。
|
||||
|
||||
const exportWikiSettlementSQL = `
|
||||
SELECT
|
||||
(SELECT COUNT(*)
|
||||
FROM knowledges AS knowledge
|
||||
WHERE knowledge.tenant_id = $1 AND knowledge.knowledge_base_id = $2 AND knowledge.deleted_at IS NULL
|
||||
AND (knowledge.pending_subtasks_count > 0 OR knowledge.parse_status IN ('pending', 'processing', 'finalizing'))),
|
||||
(SELECT COUNT(*) FROM wiki_pages AS page WHERE page.tenant_id = $1 AND page.knowledge_base_id = $2),
|
||||
(SELECT COUNT(*) FROM wiki_folders AS folder WHERE folder.tenant_id = $1 AND folder.knowledge_base_id = $2),
|
||||
(SELECT COALESCE((base.indexing_strategy ->> 'wiki_enabled')::boolean, false)
|
||||
FROM knowledge_bases AS base
|
||||
WHERE base.id = $2 AND base.tenant_id = $1 AND base.deleted_at IS NULL)`
|
||||
|
||||
const exportWikiPagesSQL = `
|
||||
SELECT to_jsonb(page)
|
||||
FROM wiki_pages AS page
|
||||
WHERE page.tenant_id = $1 AND page.knowledge_base_id = $2 AND page.deleted_at IS NULL
|
||||
ORDER BY page.slug`
|
||||
|
||||
const exportWikiFoldersSQL = `
|
||||
SELECT to_jsonb(folder)
|
||||
FROM wiki_folders AS folder
|
||||
WHERE folder.tenant_id = $1 AND folder.knowledge_base_id = $2 AND folder.deleted_at IS NULL
|
||||
ORDER BY folder.id`
|
||||
|
||||
// wikiSnapshotExporter 由 PostgresExporter 实现；文档导出器在存在该能力时把
|
||||
// 库级 wiki 快照追加进同一个 bundle，使检索与 wiki 落在同一个发布单元里。
|
||||
type wikiSnapshotExporter interface {
|
||||
ExportWiki(ctx context.Context, identity PostgresIdentity, removingKnowledgeID string) ([]TableSnapshot, error)
|
||||
}
|
||||
|
||||
// ExportWiki 导出该知识库的 wiki 快照。
|
||||
//
|
||||
// removingKnowledgeID 非空表示本次是撤权（删除）发布：剥掉该文档在 wiki 里的
|
||||
// 贡献。源端“仅停用”不会改动 wiki（WeKnora 只在删除时跑 retract 清理链），
|
||||
// 这一层剥离让“停止发布”和“源端删除”两种情况收敛到同一个结果：生产 wiki
|
||||
// 不再引用该文档。源端真删除时清理链已经跑过，剥离是幂等空操作。
|
||||
func (exporter *PostgresExporter) ExportWiki(ctx context.Context, identity PostgresIdentity, removingKnowledgeID string) ([]TableSnapshot, error) {
|
||||
if identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) {
|
||||
return nil, errors.New("invalid wiki export identity")
|
||||
}
|
||||
if removingKnowledgeID != "" && !validSearchIdentifier(removingKnowledgeID) {
|
||||
return nil, errors.New("invalid wiki removal identity")
|
||||
}
|
||||
var pending, pageCount, folderCount int
|
||||
var wikiEnabled *bool
|
||||
if err := exporter.db.QueryRow(ctx, exportWikiSettlementSQL, identity.TenantID, identity.KnowledgeBaseID).
|
||||
Scan(&pending, &pageCount, &folderCount, &wikiEnabled); err != nil {
|
||||
return nil, errors.New("read wiki snapshot state")
|
||||
}
|
||||
// 未启用 wiki 且库里没有任何 wiki 行的知识库不参与同步：包内不带 wiki 表，
|
||||
// 生产端 wiki 保持原样，避免把生产 wiki 误清空。
|
||||
if (wikiEnabled == nil || !*wikiEnabled) && pageCount == 0 && folderCount == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
// 收敛门禁：还有文档在解析或 wiki 合成中时不导出，避免产出半成品 wiki 快照。
|
||||
if pending > 0 {
|
||||
return nil, errors.New("源端 wiki 尚未收敛：该知识库仍有文档在解析或 wiki 合成中，请稍后重试")
|
||||
}
|
||||
pages, err := exporter.queryRows(ctx, exportWikiPagesSQL, identity.TenantID, identity.KnowledgeBaseID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
folders, err := exporter.queryRows(ctx, exportWikiFoldersSQL, identity.TenantID, identity.KnowledgeBaseID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, page := range pages {
|
||||
if !matchesWikiPageRow(page, identity) {
|
||||
return nil, errors.New("PostgreSQL returned a cross-scope wiki page")
|
||||
}
|
||||
}
|
||||
for _, folder := range folders {
|
||||
if !matchesWikiFolderRow(folder, identity) {
|
||||
return nil, errors.New("PostgreSQL returned a cross-scope wiki folder")
|
||||
}
|
||||
}
|
||||
if removingKnowledgeID != "" {
|
||||
pages, folders, err = exporter.retractWikiContribution(ctx, identity, removingKnowledgeID, pages, folders)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
// wiki_folders 先于 wiki_pages：页面通过 folder_id 引用文件夹。
|
||||
return []TableSnapshot{
|
||||
newTableSnapshot("wiki_folders", folders),
|
||||
newTableSnapshot("wiki_pages", pages),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// retractWikiContribution 从库级快照里剥掉一篇文档的 wiki 贡献：
|
||||
// - source_refs 形如 "<knowledge_id>|<doc_title>"，剥掉该 knowledge 的条目；
|
||||
// - chunk_refs 是该文档 chunk UUID，剥掉属于该 knowledge 的条目；
|
||||
// - 剥完没有任何来源引用的页面整页不带走（等同源端“专属页删除”）；
|
||||
// 索引页 / 日志页本来就没有 source_refs，保留；
|
||||
// - 指向被丢弃页面的 in_links / out_links 一并摘掉（避免生产死链）；
|
||||
// - 父页被丢弃时清空 parent_slug；
|
||||
// - 没有页面再引用的文件夹（以及它们的空祖先）也一并去掉，避免生产 wiki
|
||||
// 留下“空目录”这种撤权痕迹。
|
||||
func (exporter *PostgresExporter) retractWikiContribution(ctx context.Context, identity PostgresIdentity, removingKnowledgeID string, pages, folders []json.RawMessage) ([]json.RawMessage, []json.RawMessage, error) {
|
||||
chunks, err := exporter.queryRows(ctx, exportChunksSQL, removingKnowledgeID, identity.TenantID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
removedChunks := make(map[string]bool, len(chunks))
|
||||
for _, chunk := range chunks {
|
||||
chunkID, ok := matchesChunkRow(chunk, identity)
|
||||
if !ok {
|
||||
return nil, nil, errors.New("PostgreSQL returned a cross-scope chunk row")
|
||||
}
|
||||
removedChunks[chunkID] = true
|
||||
}
|
||||
kept := make([]json.RawMessage, 0, len(pages))
|
||||
droppedSlugs := map[string]bool{}
|
||||
pending := make([]map[string]any, 0, len(pages))
|
||||
keptFolderIDs := map[string]bool{}
|
||||
for _, raw := range pages {
|
||||
page, err := decodeWikiRow(raw)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
sourceRefs := stringList(page["source_refs"])
|
||||
chunkRefs := stringList(page["chunk_refs"])
|
||||
referenceBearing := len(sourceRefs) > 0 || len(chunkRefs) > 0
|
||||
keptSourceRefs := make([]string, 0, len(sourceRefs))
|
||||
for _, reference := range sourceRefs {
|
||||
if !referencesKnowledge(reference, removingKnowledgeID) {
|
||||
keptSourceRefs = append(keptSourceRefs, reference)
|
||||
}
|
||||
}
|
||||
keptChunkRefs := make([]string, 0, len(chunkRefs))
|
||||
for _, reference := range chunkRefs {
|
||||
if !removedChunks[reference] {
|
||||
keptChunkRefs = append(keptChunkRefs, reference)
|
||||
}
|
||||
}
|
||||
if referenceBearing && len(keptSourceRefs) == 0 && len(keptChunkRefs) == 0 {
|
||||
slug, _ := page["slug"].(string)
|
||||
if slug != "" {
|
||||
droppedSlugs[slug] = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
page["source_refs"] = keptSourceRefs
|
||||
page["chunk_refs"] = keptChunkRefs
|
||||
if folderID, ok := page["folder_id"].(string); ok && folderID != "" {
|
||||
keptFolderIDs[folderID] = true
|
||||
}
|
||||
pending = append(pending, page)
|
||||
}
|
||||
for _, page := range pending {
|
||||
if slug, ok := page["parent_slug"].(string); ok && slug != "" && droppedSlugs[slug] {
|
||||
page["parent_slug"] = ""
|
||||
}
|
||||
page["in_links"] = pruneSlugReferences(stringList(page["in_links"]), droppedSlugs)
|
||||
page["out_links"] = pruneSlugReferences(stringList(page["out_links"]), droppedSlugs)
|
||||
encoded, err := json.Marshal(page)
|
||||
if err != nil {
|
||||
return nil, nil, errors.New("rewrite retracted wiki page")
|
||||
}
|
||||
kept = append(kept, encoded)
|
||||
}
|
||||
return kept, pruneEmptyFolders(folders, keptFolderIDs), nil
|
||||
}
|
||||
|
||||
// pruneEmptyFolders 只保留仍被页面引用的文件夹及其祖先链。
|
||||
func pruneEmptyFolders(folders []json.RawMessage, referenced map[string]bool) []json.RawMessage {
|
||||
type folderNode struct {
|
||||
id string
|
||||
parentID string
|
||||
raw json.RawMessage
|
||||
}
|
||||
nodes := make(map[string]folderNode, len(folders))
|
||||
for _, raw := range folders {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
ParentID string `json:"parent_id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || row.ID == "" {
|
||||
continue
|
||||
}
|
||||
nodes[row.ID] = folderNode{id: row.ID, parentID: row.ParentID, raw: raw}
|
||||
}
|
||||
keep := map[string]bool{}
|
||||
for folderID := range referenced {
|
||||
for current, guard := folderID, 0; current != "" && guard < len(nodes)+1; guard++ {
|
||||
node, present := nodes[current]
|
||||
if !present || keep[current] {
|
||||
break
|
||||
}
|
||||
keep[current] = true
|
||||
current = node.parentID
|
||||
}
|
||||
}
|
||||
result := make([]json.RawMessage, 0, len(folders))
|
||||
for _, raw := range folders {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil || row.ID == "" {
|
||||
result = append(result, raw)
|
||||
continue
|
||||
}
|
||||
if keep[row.ID] {
|
||||
result = append(result, raw)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func decodeWikiRow(raw json.RawMessage) (map[string]any, error) {
|
||||
var page map[string]any
|
||||
if err := json.Unmarshal(raw, &page); err != nil {
|
||||
return nil, errors.New("invalid wiki page row")
|
||||
}
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func stringList(value any) []string {
|
||||
items, ok := value.([]any)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
result := make([]string, 0, len(items))
|
||||
for _, item := range items {
|
||||
if text, ok := item.(string); ok {
|
||||
result = append(result, text)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func referencesKnowledge(reference string, knowledgeID string) bool {
|
||||
return reference == knowledgeID || strings.HasPrefix(reference, knowledgeID+"|")
|
||||
}
|
||||
|
||||
func pruneSlugReferences(slugs []string, dropped map[string]bool) []string {
|
||||
result := make([]string, 0, len(slugs))
|
||||
for _, slug := range slugs {
|
||||
if !dropped[slug] {
|
||||
result = append(result, slug)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func matchesWikiPageRow(raw json.RawMessage, identity PostgresIdentity) bool {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
TenantID uint64 `json:"tenant_id"`
|
||||
KnowledgeBaseID string `json:"knowledge_base_id"`
|
||||
Slug string `json:"slug"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil {
|
||||
return false
|
||||
}
|
||||
return row.ID != "" && row.Slug != "" && row.TenantID == identity.TenantID && row.KnowledgeBaseID == identity.KnowledgeBaseID
|
||||
}
|
||||
|
||||
func matchesWikiFolderRow(raw json.RawMessage, identity PostgresIdentity) bool {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
TenantID uint64 `json:"tenant_id"`
|
||||
KnowledgeBaseID string `json:"knowledge_base_id"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if json.Unmarshal(raw, &row) != nil {
|
||||
return false
|
||||
}
|
||||
return row.ID != "" && row.TenantID == identity.TenantID && row.KnowledgeBaseID == identity.KnowledgeBaseID
|
||||
}
|
||||
|
||||
// hasWikiSnapshot 判断一个 bundle 是否携带库级 wiki 快照。老包（白名单里没有
|
||||
// 这两张表）不参与 wiki 同步，生产 wiki 保持原样。
|
||||
func hasWikiSnapshot(snapshot PostgresSnapshot) bool {
|
||||
return tablePresent(snapshot, "wiki_pages") && tablePresent(snapshot, "wiki_folders")
|
||||
}
|
||||
|
||||
func tablePresent(snapshot PostgresSnapshot, name string) bool {
|
||||
for _, table := range snapshot.Tables {
|
||||
if table.Name == name {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// wikiSnapshotRows 返回快照里的 wiki 行（供导入端做库级 replace 使用）。
|
||||
func wikiSnapshotRows(snapshot PostgresSnapshot) ([]TableSnapshot, bool) {
|
||||
if !hasWikiSnapshot(snapshot) {
|
||||
return nil, false
|
||||
}
|
||||
return []TableSnapshot{snapshot.Table("wiki_folders"), snapshot.Table("wiki_pages")}, true
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
type wikiStateRow struct {
|
||||
pending int
|
||||
pages int
|
||||
folders int
|
||||
wikiEnabled *bool
|
||||
err error
|
||||
}
|
||||
|
||||
func (r wikiStateRow) Scan(dest ...any) error {
|
||||
if r.err != nil {
|
||||
return r.err
|
||||
}
|
||||
*dest[0].(*int) = r.pending
|
||||
*dest[1].(*int) = r.pages
|
||||
*dest[2].(*int) = r.folders
|
||||
*dest[3].(**bool) = r.wikiEnabled
|
||||
return nil
|
||||
}
|
||||
|
||||
type wikiExportDB struct {
|
||||
state pgx.Row
|
||||
rowsByMarker map[string][]json.RawMessage
|
||||
queries []string
|
||||
}
|
||||
|
||||
func (d *wikiExportDB) QueryRow(_ context.Context, sql string, _ ...any) pgx.Row {
|
||||
d.queries = append(d.queries, sql)
|
||||
return d.state
|
||||
}
|
||||
|
||||
func (d *wikiExportDB) Query(_ context.Context, sql string, _ ...any) (pgx.Rows, error) {
|
||||
d.queries = append(d.queries, sql)
|
||||
lower := strings.ToLower(sql)
|
||||
for marker, values := range d.rowsByMarker {
|
||||
if strings.Contains(lower, marker) {
|
||||
return &jsonRows{values: values}, nil
|
||||
}
|
||||
}
|
||||
return nil, errors.New("unexpected query")
|
||||
}
|
||||
|
||||
func enabled() *bool { value := true; return &value }
|
||||
|
||||
func TestWikiExportSkipsKnowledgeBasesWithoutWiki(t *testing.T) {
|
||||
db := &wikiExportDB{state: wikiStateRow{wikiEnabled: nil}, rowsByMarker: map[string][]json.RawMessage{}}
|
||||
exporter, _ := NewPostgresExporter(db)
|
||||
tables, err := exporter.ExportWiki(context.Background(), PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"}, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tables != nil {
|
||||
t.Fatalf("tables=%#v", tables)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWikiExportBlocksUnsettledKnowledgeBase(t *testing.T) {
|
||||
db := &wikiExportDB{state: wikiStateRow{pending: 1, pages: 3, wikiEnabled: enabled()}, rowsByMarker: map[string][]json.RawMessage{}}
|
||||
exporter, _ := NewPostgresExporter(db)
|
||||
_, err := exporter.ExportWiki(context.Background(), PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"}, "")
|
||||
if err == nil || !strings.Contains(err.Error(), "未收敛") {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWikiExportCarriesPagesAndFolders(t *testing.T) {
|
||||
pages := []json.RawMessage{
|
||||
json.RawMessage(`{"id":"page-1","tenant_id":10000,"knowledge_base_id":"kb-safe","slug":"entity/alpha","title":"Alpha","source_refs":["knowledge-safe|a.md"],"in_links":[],"out_links":[]}`),
|
||||
}
|
||||
folders := []json.RawMessage{
|
||||
json.RawMessage(`{"id":"folder-1","tenant_id":10000,"knowledge_base_id":"kb-safe","parent_id":"","name":"entity"}`),
|
||||
}
|
||||
db := &wikiExportDB{
|
||||
state: wikiStateRow{pages: len(pages), folders: len(folders), wikiEnabled: enabled()},
|
||||
rowsByMarker: map[string][]json.RawMessage{
|
||||
"from wiki_pages as page": pages,
|
||||
"from wiki_folders as folder": folders,
|
||||
},
|
||||
}
|
||||
exporter, _ := NewPostgresExporter(db)
|
||||
tables, err := exporter.ExportWiki(context.Background(), PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"}, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(tables) != 2 || tables[0].Name != "wiki_folders" || tables[1].Name != "wiki_pages" {
|
||||
t.Fatalf("tables=%#v", tables)
|
||||
}
|
||||
if tables[0].RowCount != 1 || tables[1].RowCount != 1 {
|
||||
t.Fatalf("tables=%#v", tables)
|
||||
}
|
||||
for _, table := range tables {
|
||||
if newTableSnapshot(table.Name, table.Rows).SHA256 != table.SHA256 {
|
||||
t.Fatalf("digest mismatch for %s", table.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWikiExportRetractsWithdrawnDocumentContribution(t *testing.T) {
|
||||
pages := []json.RawMessage{
|
||||
json.RawMessage(`{"id":"page-shared","tenant_id":10000,"knowledge_base_id":"kb-safe","slug":"entity/shared","title":"Shared","parent_slug":"entity/exclusive","folder_id":"f-leaf","source_refs":["knowledge-safe|a.md","knowledge-other|b.md"],"chunk_refs":["chunk-safe","chunk-other"],"in_links":["entity/exclusive"],"out_links":["entity/exclusive","entity/other"]}`),
|
||||
json.RawMessage(`{"id":"page-exclusive","tenant_id":10000,"knowledge_base_id":"kb-safe","slug":"entity/exclusive","title":"Exclusive","folder_id":"f-orphan","source_refs":["knowledge-safe|a.md"],"chunk_refs":["chunk-safe"],"in_links":["entity/shared"],"out_links":[]}`),
|
||||
json.RawMessage(`{"id":"page-index","tenant_id":10000,"knowledge_base_id":"kb-safe","slug":"index","title":"Index","page_type":"index","source_refs":[],"chunk_refs":[],"in_links":[],"out_links":["entity/exclusive","entity/shared"]}`),
|
||||
}
|
||||
folders := []json.RawMessage{
|
||||
json.RawMessage(`{"id":"f-keep","tenant_id":10000,"knowledge_base_id":"kb-safe","parent_id":"","name":"entity"}`),
|
||||
json.RawMessage(`{"id":"f-mid","tenant_id":10000,"knowledge_base_id":"kb-safe","parent_id":"f-keep","name":"mid"}`),
|
||||
json.RawMessage(`{"id":"f-leaf","tenant_id":10000,"knowledge_base_id":"kb-safe","parent_id":"f-mid","name":"leaf"}`),
|
||||
json.RawMessage(`{"id":"f-orphan","tenant_id":10000,"knowledge_base_id":"kb-safe","parent_id":"","name":"orphan"}`),
|
||||
}
|
||||
db := &wikiExportDB{
|
||||
state: wikiStateRow{pages: len(pages), folders: len(folders), wikiEnabled: enabled()},
|
||||
rowsByMarker: map[string][]json.RawMessage{
|
||||
"select to_jsonb(chunk)": {json.RawMessage(`{"id":"chunk-safe","tenant_id":10000,"knowledge_id":"knowledge-safe"}`)},
|
||||
"from wiki_pages as page": pages,
|
||||
"from wiki_folders as folder": folders,
|
||||
},
|
||||
}
|
||||
exporter, _ := NewPostgresExporter(db)
|
||||
tables, err := exporter.ExportWiki(context.Background(), PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe", KnowledgeID: "knowledge-safe"}, "knowledge-safe")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var wikiPages, wikiFolders TableSnapshot
|
||||
for _, table := range tables {
|
||||
switch table.Name {
|
||||
case "wiki_pages":
|
||||
wikiPages = table
|
||||
case "wiki_folders":
|
||||
wikiFolders = table
|
||||
}
|
||||
}
|
||||
if wikiPages.RowCount != 2 {
|
||||
t.Fatalf("expected the exclusive page to be dropped: %#v", wikiPages)
|
||||
}
|
||||
rows := map[string]map[string]any{}
|
||||
for _, raw := range wikiPages.Rows {
|
||||
var row map[string]any
|
||||
if err := json.Unmarshal(raw, &row); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows[row["slug"].(string)] = row
|
||||
}
|
||||
if _, present := rows["entity/exclusive"]; present {
|
||||
t.Fatalf("exclusive page survived: %#v", rows)
|
||||
}
|
||||
if _, present := rows["index"]; !present {
|
||||
t.Fatalf("index page must survive: %#v", rows)
|
||||
}
|
||||
shared := rows["entity/shared"]
|
||||
if got := stringList(shared["source_refs"]); len(got) != 1 || got[0] != "knowledge-other|b.md" {
|
||||
t.Fatalf("source_refs=%v", got)
|
||||
}
|
||||
if got := stringList(shared["chunk_refs"]); len(got) != 1 || got[0] != "chunk-other" {
|
||||
t.Fatalf("chunk_refs=%v", got)
|
||||
}
|
||||
if got := stringList(shared["out_links"]); len(got) != 1 || got[0] != "entity/other" {
|
||||
t.Fatalf("out_links=%v", got)
|
||||
}
|
||||
if got := stringList(shared["in_links"]); len(got) != 0 {
|
||||
t.Fatalf("in_links=%v", got)
|
||||
}
|
||||
if parent, _ := shared["parent_slug"].(string); parent != "" {
|
||||
t.Fatalf("parent_slug=%q", parent)
|
||||
}
|
||||
indexRow := rows["index"]
|
||||
if got := stringList(indexRow["out_links"]); len(got) != 1 || got[0] != "entity/shared" {
|
||||
t.Fatalf("index out_links=%v", got)
|
||||
}
|
||||
keptFolders := map[string]bool{}
|
||||
for _, raw := range wikiFolders.Rows {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &row); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keptFolders[row.ID] = true
|
||||
}
|
||||
if wikiFolders.RowCount != 3 {
|
||||
t.Fatalf("expected only the referenced folder chain to survive: %#v", keptFolders)
|
||||
}
|
||||
for _, want := range []string{"f-keep", "f-mid", "f-leaf"} {
|
||||
if !keptFolders[want] {
|
||||
t.Fatalf("folder %s must survive: %v", want, keptFolders)
|
||||
}
|
||||
}
|
||||
if keptFolders["f-orphan"] {
|
||||
t.Fatalf("folder without surviving pages leaked into production: %v", keptFolders)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWikiExportRejectsCrossScopePage(t *testing.T) {
|
||||
db := &wikiExportDB{
|
||||
state: wikiStateRow{pages: 1, wikiEnabled: enabled()},
|
||||
rowsByMarker: map[string][]json.RawMessage{
|
||||
"from wiki_pages as page": {json.RawMessage(`{"id":"page-1","tenant_id":10000,"knowledge_base_id":"kb-other","slug":"entity/x","title":"X"}`)},
|
||||
"from wiki_folders as folder": {},
|
||||
},
|
||||
}
|
||||
exporter, _ := NewPostgresExporter(db)
|
||||
_, err := exporter.ExportWiki(context.Background(), PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"}, "")
|
||||
if err == nil || !strings.Contains(err.Error(), "cross-scope") {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "kb-other") {
|
||||
t.Fatalf("error leaked scope value: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
#!/usr/bin/env bash
|
||||
# 新建 wiki 知识库 验收脚本：全新 KB（源/目标同 ID）+ 单文档导入 → 发布 → 撤权
|
||||
# 用法：bash scripts/wiki_e2e_newkb.sh
|
||||
set -uo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
REPO=$PWD
|
||||
SRC=/tmp/wk_curl.cfg
|
||||
CLONE_FROM=c12ff4d5-dd87-423b-ab0f-af745b9d54a7
|
||||
NEW_KB=$(python3 -c "import uuid;print(uuid.uuid4())")
|
||||
DIST=$REPO/dist/release-channel-e2e
|
||||
echo "新知识库 ID: $NEW_KB"
|
||||
|
||||
src_sql() { docker exec WKNFull-postgres psql -U weknora -d weknora_full -At -P pager=off -c "$1"; }
|
||||
prod_sql() { docker run --rm --network host --env-file /tmp/target_pg.env postgres:16-alpine psql -At -P pager=off -c "$1"; }
|
||||
|
||||
echo "### 1. 源端新建 wiki 知识库（克隆同构 KB 行，开 wiki）"
|
||||
src_sql "create temp table t as select * from knowledge_bases where id='$CLONE_FROM';
|
||||
update t set id='$NEW_KB', name='wiki-e2e-fresh', created_at=now(), updated_at=now(),
|
||||
indexing_strategy = indexing_strategy || '{\"wiki_enabled\": true}'::jsonb;
|
||||
insert into knowledge_bases select * from t;"
|
||||
echo "源端新库: $(src_sql "select id||' | '||name||' | wiki='||(indexing_strategy->>'wiki_enabled') from knowledge_bases where id='$NEW_KB'")"
|
||||
|
||||
echo "### 2. 目标端新建同 ID 库（克隆目标已有的同构 KB 行，开 wiki）"
|
||||
prod_sql "create temp table t as select * from knowledge_bases where id='$CLONE_FROM';
|
||||
update t set id='$NEW_KB', name='wiki-e2e-fresh', created_at=now(), updated_at=now(),
|
||||
indexing_strategy = indexing_strategy || '{\"wiki_enabled\": true}'::jsonb;
|
||||
insert into knowledge_bases select * from t;"
|
||||
echo "目标端新库: $(prod_sql "select id||' | '||name||' | wiki='||coalesce(indexing_strategy->>'wiki_enabled','?') from knowledge_bases where id='$NEW_KB'")"
|
||||
|
||||
echo "### 3. 生成该库专用发布路由配置（同目标端点，仅换 kb）"
|
||||
python3 - "$NEW_KB" <<'PY'
|
||||
import json,sys
|
||||
kb=sys.argv[1]
|
||||
cfg=json.load(open('config/release.sim.json'))
|
||||
cfg['routes'][0]['knowledge_base_id']=kb
|
||||
json.dump(cfg,open('config/release.wiki-e2e.json','w'),ensure_ascii=False,indent=1)
|
||||
print('写入 config/release.wiki-e2e.json kb='+kb)
|
||||
PY
|
||||
chmod 600 config/release.wiki-e2e.json
|
||||
|
||||
echo "### 4. 导入单文档并等 wiki 收敛"
|
||||
cat > /tmp/wiki_e2e_fresh.md <<'EOF'
|
||||
# 新建 wiki 知识库验证文档
|
||||
|
||||
本知识库用于验证“新建 wiki 知识库后，单文档的检索与 wiki 是否一起搬运到生产”。
|
||||
|
||||
知识治理平台把测试端 WeKnora 的文档发布到生产端 WeKnora。
|
||||
本次验证关注两条轨道：检索（PostgreSQL 关系行 + Elasticsearch 分片）与
|
||||
wiki 快照（wiki_pages 与 wiki_folders 两张库级表）。
|
||||
|
||||
撤销发布时，生产端应当同时去掉检索与 wiki 里这篇文档的贡献。
|
||||
EOF
|
||||
DOC=$(curl -s -K $SRC -F "file=@/tmp/wiki_e2e_fresh.md" -F "fileName=wiki-e2e-fresh.md" \
|
||||
"http://127.0.0.1:8080/api/v1/knowledge-bases/$NEW_KB/knowledge/file" | python3 -c "import sys,json;print(json.load(sys.stdin)['data']['id'])")
|
||||
echo "文档 ID: $DOC"; printf '%s' "$DOC" > /tmp/wiki_e2e_fresh_id
|
||||
for i in $(seq 1 30); do
|
||||
st=$(src_sql "select parse_status||'/'||pending_subtasks_count from knowledges where id='$DOC'")
|
||||
pages=$(src_sql "select count(*) from wiki_pages where knowledge_base_id='$NEW_KB' and deleted_at is null")
|
||||
echo " [$i] $st wiki_pages=$pages"
|
||||
case "$st" in completed/0) break;; esac
|
||||
sleep 15
|
||||
done
|
||||
|
||||
release() {
|
||||
docker run --rm --user "$(id -u):$(id -g)" \
|
||||
--network knowledge-governance-p1_default --network weknora071full-network \
|
||||
--add-host prod-pg:host-gateway --add-host prod-es:host-gateway --add-host prod-weknora:host-gateway \
|
||||
-v "$REPO/secrets/release-sim":/run/release-secrets:ro \
|
||||
-v "$REPO/config/release.wiki-e2e.json":/run/release-sim/release.wiki-e2e.json:ro \
|
||||
-v knowledge-governance-release-bundles:/var/lib/knowledge-governance/release-bundles \
|
||||
-v "$DIST":/release-channel-e2e:ro \
|
||||
--entrypoint /release-channel-e2e knowledge-governance-p1-release-worker:latest \
|
||||
-config /run/release-sim/release.wiki-e2e.json -knowledge "$1" -mode "$2" -revision "$3" 2>&1 | grep -E "STEP|bundle|wiki (before|after)"
|
||||
}
|
||||
|
||||
echo "### 5. 发布（新建库）"
|
||||
# 归档根目录必须预先存在且为 0700（NewBundleArchive 只校验不创建）
|
||||
docker run --rm --user "$(id -u):$(id -g)" -v knowledge-governance-release-bundles:/bundles alpine:3.20 \
|
||||
sh -c "mkdir -p /bundles/$NEW_KB && chmod 700 /bundles/$NEW_KB"
|
||||
release "$DOC" publish wiki-e2e-newkb-v1
|
||||
echo "--- 生产端证据"
|
||||
prod_sql "select 'knowledges='||count(*) filter (where enable_status='enabled') from knowledges where knowledge_base_id='$NEW_KB'"
|
||||
prod_sql "select 'wiki_pages='||count(*) filter (where deleted_at is null) from wiki_pages where knowledge_base_id='$NEW_KB'"
|
||||
|
||||
echo "### 6. 撤权（新建库）"
|
||||
release "$DOC" withdraw wiki-e2e-newkb-v2
|
||||
echo "--- 生产端证据（撤权后）"
|
||||
prod_sql "select id||' enable='||enable_status from knowledges where knowledge_base_id='$NEW_KB'"
|
||||
prod_sql "select 'wiki_pages='||count(*)||' refs_still_citing='||count(*) filter (where source_refs::text like '%$DOC%') from wiki_pages where knowledge_base_id='$NEW_KB'"
|
||||
curl -s "http://172.17.0.1:29200/weknora_test_chunks/_count" -H 'Content-Type: application/json' \
|
||||
-d "{\"query\":{\"bool\":{\"filter\":[{\"term\":{\"knowledge_id.keyword\":\"$DOC\"}},{\"term\":{\"is_enabled\":true}}]}}}" \
|
||||
| python3 -c "import sys,json;print('ES enabled chunks:',json.load(sys.stdin).get('count'))"
|
||||
@@ -0,0 +1,107 @@
|
||||
#!/usr/bin/env bash
|
||||
# wiki 同搬通道 5 文档验收脚本（本地模拟生产 = 测试服 weknora-test-v071）
|
||||
# 用法：bash scripts/wiki_e2e_test5.sh
|
||||
# 前置：三条 ssh 隧道（172.17.0.1:25432/29200/28080）已起，dist/release-channel-e2e 已构建
|
||||
set -uo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
REPO=$PWD
|
||||
KB=c12ff4d5-dd87-423b-ab0f-af745b9d54a7
|
||||
WK=http://127.0.0.1:8080
|
||||
CFG=/run/release-sim/release.sim.json
|
||||
DIST=$REPO/dist/release-channel-e2e
|
||||
|
||||
api() { curl -s -K /tmp/wk_curl.cfg "$@"; }
|
||||
prod_sql() { docker run --rm --network host --env-file /tmp/target_pg.env postgres:16-alpine psql -At -P pager=off -c "$1"; }
|
||||
prod_es_count() { # 用 keyword 子字段精确匹配（knowledge_id 是 text 字段，UUID 会被分词）
|
||||
curl -s "http://172.17.0.1:29200/weknora_test_chunks/_count" -H 'Content-Type: application/json' \
|
||||
-d "{\"query\":{\"bool\":{\"filter\":[{\"term\":{\"knowledge_id.keyword\":\"$1\"}},{\"term\":{\"is_enabled\":true}}]}}}" \
|
||||
| python3 -c "import sys,json;print(json.load(sys.stdin).get('count'))"
|
||||
}
|
||||
release() { # release <knowledge> <mode> <revision>
|
||||
docker run --rm --user "$(id -u):$(id -g)" \
|
||||
--network knowledge-governance-p1_default --network weknora071full-network \
|
||||
--add-host prod-pg:host-gateway --add-host prod-es:host-gateway --add-host prod-weknora:host-gateway \
|
||||
-v "$REPO/secrets/release-sim":/run/release-secrets:ro \
|
||||
-v "$REPO/config/release.sim.json":/run/release-sim/release.sim.json:ro \
|
||||
-v knowledge-governance-release-bundles:/var/lib/knowledge-governance/release-bundles \
|
||||
-v "$DIST":/release-channel-e2e:ro \
|
||||
--entrypoint /release-channel-e2e knowledge-governance-p1-release-worker:latest \
|
||||
-config "$CFG" -knowledge "$1" -mode "$2" -revision "$3" 2>&1 | grep -E "STEP|bundle|wiki (before|after)|route kb"
|
||||
}
|
||||
|
||||
echo "### 1. 上传 4 篇同主题文档（与首篇共贡 wiki 页）"
|
||||
write_doc() { cat > "/tmp/wiki_e2e_$1.md"; }
|
||||
write_doc b <<'EOF'
|
||||
# 来源与目录：停止知识的语义
|
||||
|
||||
知识治理平台在“来源与目录”视图里可以停止某条知识。停止的含义是：生产端不再检索它。
|
||||
需要注意，停止不会改动测试端 WeKnora 的 wiki：只有删除文档才会触发 wiki 清理链。
|
||||
因此 wiki 快照在撤权时由发布通道负责剥离该文档的贡献。
|
||||
EOF
|
||||
write_doc c <<'EOF'
|
||||
# 发布通道之 PostgreSQL 与 Elasticsearch
|
||||
|
||||
发布通道把知识治理平台评审通过的知识写进生产端 WeKnora。
|
||||
PostgreSQL 负责关系行：knowledges、chunks、knowledge_tags、resources。
|
||||
Elasticsearch 负责检索分片与向量，索引名称为 weknora_test_chunks。
|
||||
撤权时 Elasticsearch 通过 delete-by-query 清空该文档分片。
|
||||
EOF
|
||||
write_doc d <<'EOF'
|
||||
# 发布通道之 OSS 对象与图片资源
|
||||
|
||||
文档里的图片与附件存放在对象存储 OSS。发布时先写入暂存路径，再用 CopyObject 拷到生产桶。
|
||||
resources 表保存 storage:// 定位符，搬运时只替换桶名。
|
||||
知识治理平台与测试端 WeKnora、生产端 WeKnora 共用同一套对象通道实现。
|
||||
EOF
|
||||
write_doc e <<'EOF'
|
||||
# 撤权与回滚演练
|
||||
|
||||
撤权（withdraw）会把生产端知识置为 disabled 并清空检索分片，同时重放 wiki 快照。
|
||||
回滚（rollback）会整包还原上一版 revision，包括 wiki 快照。
|
||||
因此生产端 WeKnora 永远不会出现“检索没了、wiki 还在”的中间态。
|
||||
EOF
|
||||
for k in b c d e; do
|
||||
id=$(api -X POST -F "file=@/tmp/wiki_e2e_$k.md" -F "fileName=wiki-e2e-$k.md" "$WK/api/v1/knowledge-bases/$KB/knowledge/file" | python3 -c "import sys,json;print(json.load(sys.stdin)['data']['id'])")
|
||||
echo "doc $k -> $id"
|
||||
printf '%s\n' "$id" > "/tmp/wiki_e2e_id_$k"
|
||||
done
|
||||
|
||||
echo "### 2. 等 5 篇全部收敛（parse=completed 且 wiki 子任务为 0）"
|
||||
for i in $(seq 1 40); do
|
||||
pending=$(docker exec WKNFull-postgres psql -U weknora -d weknora_full -At -c "select count(*) from knowledges where knowledge_base_id='$KB' and deleted_at is null and (parse_status <> 'completed' or pending_subtasks_count > 0)")
|
||||
total=$(docker exec WKNFull-postgres psql -U weknora -d weknora_full -At -c "select count(*) from knowledges where knowledge_base_id='$KB' and deleted_at is null")
|
||||
pages=$(docker exec WKNFull-postgres psql -U weknora -d weknora_full -At -c "select count(*) from wiki_pages where knowledge_base_id='$KB' and deleted_at is null")
|
||||
echo " [$i] docs=$total 未收敛=$pending wiki_pages=$pages"
|
||||
[ "$total" -ge 5 ] && [ "$pending" = "0" ] && break
|
||||
sleep 20
|
||||
done
|
||||
|
||||
echo "### 3. 逐篇发布（每次发布会推整库 wiki 快照）"
|
||||
i=0
|
||||
for id in $(cat /tmp/wiki_e2e_id_b) 9be3e0fa-e699-4a8e-be0c-0d934ee33294 $(cat /tmp/wiki_e2e_id_c) $(cat /tmp/wiki_e2e_id_d) $(cat /tmp/wiki_e2e_id_e); do
|
||||
i=$((i+1))
|
||||
echo "--- publish #$i $id"
|
||||
release "$id" publish "wiki-e2e-5doc-p$i" | tail -4
|
||||
done
|
||||
|
||||
echo "### 4. 生产端证据（5 篇）"
|
||||
prod_sql "select 'enabled='||count(*) filter (where enable_status='enabled')||' total='||count(*) from knowledges where knowledge_base_id='$KB'"
|
||||
prod_sql "select 'wiki_pages='||(select count(*) from wiki_pages where knowledge_base_id='$KB')||' wiki_folders='||(select count(*) from wiki_folders where knowledge_base_id='$KB')"
|
||||
echo "共贡页面（source_refs 多于一条）:"
|
||||
prod_sql "select slug||' <- '||source_refs::text from wiki_pages where knowledge_base_id='$KB' and jsonb_array_length(coalesce(source_refs,'[]'::jsonb)) > 1 order by slug limit 8"
|
||||
for id in $(cat /tmp/wiki_e2e_id_b) $(cat /tmp/wiki_e2e_id_c) 9be3e0fa-e699-4a8e-be0c-0d934ee33294; do
|
||||
echo " ES $id count=$(prod_es_count "$id")"
|
||||
done
|
||||
|
||||
echo "### 5. 撤权其中一篇（doc b），验证共贡页面保留、专属页面消失、其它文档不受影响"
|
||||
release "$(cat /tmp/wiki_e2e_id_b)" withdraw "wiki-e2e-5doc-w1" | tail -5
|
||||
echo "--- 生产端证据（撤权后）"
|
||||
prod_sql "select id||' enable='||enable_status from knowledges where knowledge_base_id='$KB' order by enable_status, id"
|
||||
prod_sql "select 'wiki_pages='||count(*) from wiki_pages where knowledge_base_id='$KB'"
|
||||
prod_sql "select 'folders='||count(*) from wiki_folders where knowledge_base_id='$KB'"
|
||||
echo "还有引用被撤权文档的页面（应为 0）:"
|
||||
prod_sql "select count(*) from wiki_pages where knowledge_base_id='$KB' and source_refs::text like '%$(cat /tmp/wiki_e2e_id_b)%'"
|
||||
echo "共贡页面是否仍在:"
|
||||
prod_sql "select slug||' <- '||source_refs::text from wiki_pages where knowledge_base_id='$KB' and jsonb_array_length(coalesce(source_refs,'[]'::jsonb)) > 0 order by slug limit 8"
|
||||
echo "ES 分片:"
|
||||
for id in $(cat /tmp/wiki_e2e_id_b) $(cat /tmp/wiki_e2e_id_c); do echo " $id count=$(prod_es_count "$id")"; done
|
||||
Reference in New Issue
Block a user