From 0d9091e4644cf3eb053fe2c852bdbab90a51cc24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Wed, 4 Nov 2026 12:40:12 +0800 Subject: [PATCH] =?UTF-8?q?fix(library):=20=E7=9B=AE=E6=A0=87=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=E6=98=A0=E5=B0=84=E5=85=BC=E5=AE=B9=E6=97=A7=E8=AE=A1?= =?UTF-8?q?=E5=88=92=E4=B8=8E=E7=9C=9F=E5=AE=9E=E5=9B=9E=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 空 mode 计划不映射；索引 source_id 按语义映射；relation 按字段表解析。 隔离回归按持久映射 ID 断言，DS02 故障注入真实命中。 --- .../v071/data_closure_r10_test.go | 12 +- .../v071/data_contract_r10_test.go | 8 +- .../v071/data_matrix_r11_test.go | 61 +++++++- .../v071/data_schema_r11_test.go | 4 +- .../v071/data_state_r12_test.go | 20 +-- .../releaseadapter/v071/ds02_oracle_test.go | 70 +++++++--- ...library_chunk_sequence_integration_test.go | 2 +- .../v071/library_index_recovery_gaps_test.go | 6 +- ...rary_production_target_integration_test.go | 15 +- .../releaseadapter/v071/library_target_ids.go | 102 +++++++++----- .../library_target_ids_integration_test.go | 56 ++++++++ .../v071/library_target_ids_test.go | 17 ++- .../v071/library_target_path.go | 12 +- .../v071/library_verify_content.go | 3 + .../v071/target_id_expect_test.go | 131 ++++++++++++++++++ 15 files changed, 428 insertions(+), 91 deletions(-) create mode 100644 internal/releaseadapter/v071/target_id_expect_test.go diff --git a/internal/releaseadapter/v071/data_closure_r10_test.go b/internal/releaseadapter/v071/data_closure_r10_test.go index e562d13..cb498dd 100644 --- a/internal/releaseadapter/v071/data_closure_r10_test.go +++ b/internal/releaseadapter/v071/data_closure_r10_test.go @@ -93,16 +93,20 @@ func TestR10IncomingRelationClosureIsolated(t *testing.T) { if result.State != "succeeded" { t.Fatalf("nonempty tuple publication: %+v", result) } - wantID := "doc-src:tag-src" + wantID := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src") + ":" + + expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src") if mode == "valid-existing" { - wantID = "doc-target:tag-target" + wantID = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-target") + ":" + + expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-target") } found := false for _, row := range result.Backup.WriteSet.Rows { if row.Table == "knowledge_tag_relations" && row.ID == wantID { found = true - if row.Existing != (mode == "valid-existing") { - t.Fatal("tuple existence misclassified", row) + // 映射后所有 tuple 都是目标新身份；即使源 ID 恰是目标库已存在的 + // doc-target/tag-target，它们会被派生成本次的新 ID，写前不存在。 + if row.Existing { + t.Fatal("mapped tuple misclassified as pre-existing", row) } } } diff --git a/internal/releaseadapter/v071/data_contract_r10_test.go b/internal/releaseadapter/v071/data_contract_r10_test.go index 164e24b..6391f8b 100644 --- a/internal/releaseadapter/v071/data_contract_r10_test.go +++ b/internal/releaseadapter/v071/data_contract_r10_test.go @@ -171,13 +171,17 @@ func TestR10SequenceOrderingReplayIsolated(t *testing.T) { if e := fx.pool.QueryRow(fx.ctx, `SELECT array_agg(id ORDER BY chunk_index ASC) FROM chunks WHERE knowledge_base_id='kb-target'`).Scan(&chunkOrder); e != nil { t.Fatal(e) } - if !reflect.DeepEqual(tagOrder, []string{"tag-src", "tag-src-2"}) || !reflect.DeepEqual(chunkOrder, []string{"chunk-src", "chunk-src-2"}) { + wantTags := []string{expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src"), + expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src-2")} + wantChunks := []string{expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src"), + expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src-2")} + if !reflect.DeepEqual(tagOrder, wantTags) || !reflect.DeepEqual(chunkOrder, wantChunks) { t.Fatal("application sort changed after numeric remapping", tagOrder, chunkOrder) } for _, m := range result.Backup.WriteSet.LocalIDs { if m.Table == "chunks" { var id string - if e := fx.pool.QueryRow(fx.ctx, `SELECT id FROM chunks WHERE tenant_id=$1 AND seq_id=$2::bigint`, wsTargetTenant, m.Target).Scan(&id); e != nil || (m.Source == "100" && id != "chunk-src") || (m.Source == "2" && id != "chunk-src-2") { + if e := fx.pool.QueryRow(fx.ctx, `SELECT id FROM chunks WHERE tenant_id=$1 AND seq_id=$2::bigint`, wsTargetTenant, m.Target).Scan(&id); e != nil || (m.Source == "100" && id != expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")) || (m.Source == "2" && id != expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src-2")) { t.Fatal("seq-ID consumer disagrees with frozen map", e) } } diff --git a/internal/releaseadapter/v071/data_matrix_r11_test.go b/internal/releaseadapter/v071/data_matrix_r11_test.go index bc3df05..d79733b 100644 --- a/internal/releaseadapter/v071/data_matrix_r11_test.go +++ b/internal/releaseadapter/v071/data_matrix_r11_test.go @@ -221,7 +221,7 @@ func r11Source(t *testing.T, fx *writeSetFixture, tweak func(map[string][]map[st } }) } -func r11Reject(t *testing.T, fx *writeSetFixture, snapshot string) { +func r11RejectNatural(t *testing.T, fx *writeSetFixture, snapshot string) { t.Helper() before := r10State(t, fx) target := fx.target(t) @@ -232,6 +232,60 @@ func r11Reject(t *testing.T, fx *writeSetFixture, snapshot string) { } r10Unchanged(t, fx, before) } + +func r11Reject(t *testing.T, fx *writeSetFixture, snapshot, table, sourceID string) { + t.Helper() + before := r10State(t, fx) + var beforeRow string + if e := fx.pool.QueryRow(fx.ctx, `SELECT to_jsonb(x)::text FROM `+table+` AS x WHERE id=$1`, sourceID).Scan(&beforeRow); e != nil { + t.Fatalf("read foreign row before publish: %v", e) + } + beforeState := map[string]string{"row:" + table: beforeRow} + target := fx.target(t) + out := libraryrelease.Run(fx.ctx, requestFor(snapshot), target, fx.backups(t, target), nil) + r10Save(t, "outcome", out) + if table == "resources" { + // 资源是共享字节，不参与 UUID 映射：同 ID 属于他库必须继续拒绝。 + if out.State != "failed" || out.FailedAt != "" { + t.Errorf("shared resource id collision must be rejected pre-stage: %+v", out) + } + r10Unchanged(t, fx, before) + return + } + if out.State != "succeeded" { + t.Fatalf("mapped identity publish should succeed: %+v", out) + } + // 旧库原行必须保留，且计划里不得再出现源 ID。 + for _, row := range out.Backup.WriteSet.Rows { + if row.Table == table && row.ID == sourceID { + t.Fatalf("writeset still carries source identity %s/%s", table, sourceID) + } + } + mapped := expectedTargetID(wsTargetTenant, wsTargetKB, table, sourceID) + var foreign, targetRows int + if e := fx.pool.QueryRow(fx.ctx, `SELECT count(*) FROM `+table+` WHERE id=$1`, sourceID).Scan(&foreign); e != nil || foreign != 1 { + t.Fatalf("foreign row %s/%s changed or lost: %d %v", table, sourceID, foreign, e) + } + // 旧库行必须逐列字节不变（不只是"还在"）。 + beforeRow, afterRow := "", "" + if e := fx.pool.QueryRow(fx.ctx, `SELECT to_jsonb(x)::text FROM `+table+` AS x WHERE id=$1`, sourceID).Scan(&afterRow); e != nil { + t.Fatalf("read foreign row %s: %v", table, e) + } + beforeRow = beforeState["row:"+table] + if beforeRow != "" && beforeRow != afterRow { + t.Fatalf("foreign row %s/%s content changed", table, sourceID) + } + if e := fx.pool.QueryRow(fx.ctx, `SELECT count(*) FROM `+table+` WHERE id=$1`, mapped).Scan(&targetRows); e != nil || targetRows != 1 { + t.Fatalf("mapped row %s/%s missing: %d %v", table, mapped, targetRows, e) + } + if tablesWithKB[table] { + var tenant int64 + var kb string + if e := fx.pool.QueryRow(fx.ctx, `SELECT tenant_id, knowledge_base_id FROM `+table+` WHERE id=$1`, mapped).Scan(&tenant, &kb); e != nil || tenant != int64(wsTargetTenant) || kb != wsTargetKB { + t.Fatalf("mapped row %s landed at %d/%s", table, tenant, kb) + } + } +} func TestR11StableIdentityMatrixIsolated(t *testing.T) { for _, table := range []string{"knowledges", "chunks", "knowledge_tags", "resources", "resource_bindings", "wiki_folders", "wiki_pages", "wiki_page_issues"} { for _, foreign := range []string{"same-tenant-other-kb", "other-tenant"} { @@ -250,7 +304,6 @@ func TestR11StableIdentityMatrixIsolated(t *testing.T) { switch table { case "chunks": row["knowledge_id"] = "doc-other" - row["seq_id"] = int64(900000001) case "resources": delete(row, "knowledge_base_id") row["handle"] = strings.Repeat("x", 22) @@ -268,7 +321,7 @@ func TestR11StableIdentityMatrixIsolated(t *testing.T) { } r11Insert(t, fx, table, row) snapshot := r11Source(t, fx, nil) - r11Reject(t, fx, snapshot) + r11Reject(t, fx, snapshot, table, id) }) } } @@ -328,7 +381,7 @@ func TestR11NaturalKeyMatrixIsolated(t *testing.T) { p[table] = append(p[table], cp) }) if mode != "valid" && !strings.HasSuffix(mode, "positive") { - r11Reject(t, fx, snapshot) + r11RejectNatural(t, fx, snapshot) return } before := r10State(t, fx) diff --git a/internal/releaseadapter/v071/data_schema_r11_test.go b/internal/releaseadapter/v071/data_schema_r11_test.go index e0bf1b0..fa04403 100644 --- a/internal/releaseadapter/v071/data_schema_r11_test.go +++ b/internal/releaseadapter/v071/data_schema_r11_test.go @@ -97,7 +97,7 @@ func TestR11CatalogBindingIsolated(t *testing.T) { "trigger": `CREATE FUNCTION r11_noop() RETURNS trigger LANGUAGE plpgsql AS 'BEGIN RETURN NEW; END';CREATE TRIGGER r11_noop BEFORE INSERT ON wiki_pages FOR EACH ROW EXECUTE FUNCTION r11_noop()`, } r10Exec(t, fx, statements[mode]) - r11Reject(t, fx, snapshot) + r11RejectNatural(t, fx, snapshot) // An identically extended source/target must not make an unreviewed FK safe. if strings.Contains(mode, "fk") { now, _, e := target.migrationCatalog(fx.ctx) @@ -157,7 +157,7 @@ func TestR11ReferenceAndCursorIsolated(t *testing.T) { } }) if mode != "valid" && mode != "linked-valid" && mode != "legacy-pages" && mode != "historical-log" { - r11Reject(t, fx, snapshot) + r11RejectNatural(t, fx, snapshot) return } target := fx.target(t) diff --git a/internal/releaseadapter/v071/data_state_r12_test.go b/internal/releaseadapter/v071/data_state_r12_test.go index 5368f35..8a995c8 100644 --- a/internal/releaseadapter/v071/data_state_r12_test.go +++ b/internal/releaseadapter/v071/data_state_r12_test.go @@ -23,10 +23,10 @@ func r12ESDoc(t *testing.T, fx *writeSetFixture, id string) map[string]any { return recoveryIndexReadback(t, fx)[id] } -func r12PutDoc(t *testing.T, fx *writeSetFixture, id string, tenant uint64, kb, knowledge, content string, embedding []float64) { +func r12PutDoc(t *testing.T, fx *writeSetFixture, id, chunkID string, tenant uint64, kb, knowledge, content string, embedding []float64) { t.Helper() body, err := json.Marshal(map[string]any{"tenant_id": tenant, "knowledge_base_id": kb, - "knowledge_id": knowledge, "chunk_id": id, "source_id": id, "content": content, "embedding": embedding}) + "knowledge_id": knowledge, "chunk_id": chunkID, "source_id": chunkID, "content": content, "embedding": embedding}) if err != nil { t.Fatal(err) } @@ -158,7 +158,9 @@ func TestR12IndexOracleDetectsContentVectorAndMetadataChangesIsolated(t *testing if err := target.Verify(fx.ctx, expect); err != nil { t.Fatalf("baseline oracle must pass: %v", err) } - published := r12ESDoc(t, fx, "chunk-src") + docID := expectedIndexID("chunk-src") + chunkID := expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src") + published := r12ESDoc(t, fx, docID) if published == nil { t.Fatal("published document missing") } @@ -168,29 +170,29 @@ func TestR12IndexOracleDetectsContentVectorAndMetadataChangesIsolated(t *testing originalKnowledge, _ := source["knowledge_id"].(string) // 1) 正文被改。 - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, "被篡改的正文", []float64{0.1, 0.2}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, "被篡改的正文", []float64{0.1, 0.2}) if err := target.Verify(fx.ctx, expect); err == nil { t.Fatal("正文篡改没有让独立 oracle 变红") } - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) if err := target.Verify(fx.ctx, expect); err != nil { t.Fatalf("恢复正文后 oracle 必须通过：%v", err) } // 2) 向量被改（同维不同值）。 - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.9, 0.8}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.9, 0.8}) if err := target.Verify(fx.ctx, expect); err == nil { t.Fatal("向量篡改没有让独立 oracle 变红") } - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) if err := target.Verify(fx.ctx, expect); err != nil { t.Fatalf("恢复向量后 oracle 必须通过：%v", err) } // 3) metadata（knowledge_id）被改。 - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, "doc-target", originalContent, []float64{0.1, 0.2}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, "doc-target", originalContent, []float64{0.1, 0.2}) if err := target.Verify(fx.ctx, expect); err == nil { t.Fatal("metadata 篡改没有让独立 oracle 变红") } - r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) + r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2}) if err := target.Verify(fx.ctx, expect); err != nil { t.Fatalf("恢复 metadata 后 oracle 必须通过：%v", err) } diff --git a/internal/releaseadapter/v071/ds02_oracle_test.go b/internal/releaseadapter/v071/ds02_oracle_test.go index cadaf4d..10195fa 100644 --- a/internal/releaseadapter/v071/ds02_oracle_test.go +++ b/internal/releaseadapter/v071/ds02_oracle_test.go @@ -903,7 +903,8 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st diffs = append(diffs, fmt.Sprintf("index docs: want %d got %d", len(spec.rows["chunks"])+1, len(docs))) } for _, chunk := range spec.rows["chunks"] { - id := fmt.Sprint(chunk["id"]) + sourceID := fmt.Sprint(chunk["id"]) + id := expectedIndexID(sourceID) doc, ok := docs[id] if !ok { diffs = append(diffs, "index doc missing: "+id) @@ -911,8 +912,11 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st } source := doc want := map[string]any{"tenant_id": json.Number(strconv.FormatUint(wsTargetTenant, 10)), - "knowledge_base_id": wsTargetKB, "knowledge_id": chunk["knowledge_id"], "chunk_id": id, - "source_id": id, "content": chunk["content"], "embedding": spec.vectors[id]} + "knowledge_base_id": wsTargetKB, + "knowledge_id": expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", fmt.Sprint(chunk["knowledge_id"])), + "chunk_id": expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", sourceID), + "source_id": expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", sourceID), + "content": chunk["content"], "embedding": spec.vectors[sourceID]} for key, value := range want { if fmt.Sprint(source[key]) != fmt.Sprint(value) { diffs = append(diffs, fmt.Sprintf("index doc %s.%s: want %v got %v", id, key, value, source[key])) @@ -971,6 +975,9 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st } copied["tenant_id"] = json.Number(strconv.FormatUint(wsTargetTenant, 10)) copied["knowledge_base_id"] = wsTargetKB + if knowledge, ok := copied["knowledge_id"].(string); ok && knowledge != "" { + copied["knowledge_id"] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", knowledge) + } raw, _ := json.Marshal(copied) wantTuples = append(wantTuples, string(raw)) } @@ -1103,6 +1110,18 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) { spec := ds02SourceSpec(t, fx) expectation := ds02Expected(t, fx, spec) expected, paths := expectation.Rows, expectation.Paths + // 独立 oracle：只对"来源快照将写进目标范围"的期望行按 scoped-sha256-v1 + // 推导目标身份；随后追加的同租户/外库保留行仍是原身份，不能映射。 + for table := range expected { + for i, row := range expected[table] { + expected[table][i] = expectedTargetRow(table, row) + } + } + for table, bySeq := range expectation.RowsBySource { + for seq, row := range bySeq { + bySeq[seq] = expectedTargetRow(table, row) + } + } for _, table := range []string{"resources", "resource_bindings"} { expected[table] = append(expected[table], ds02SameTenantRows(t, fx, table)...) } @@ -1112,6 +1131,7 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) { ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM knowledge_tag_relations AS x WHERE knowledge_id NOT IN (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2)`, wsTargetTenant, wsTargetKB)...) + sequencesBefore := ds02SequenceSummary(t, fx) snapshot := ds02Seal(t, fx, spec) recorder, inventory := ds02AttachEgress(t, fx) @@ -1282,8 +1302,28 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) { foreignPath := wsLocator("backend-target", "minio", fx.bucket, 10001, "res-other") foreignDoc := `{"tenant_id":20000,"knowledge_base_id":"kb-foreign","knowledge_id":"doc-foreign","chunk_id":"chunk-foreign","source_id":"chunk-foreign","content":"外部正文","embedding":[0.1,0.2]}` foreignDocTampered := `{"tenant_id":20000,"knowledge_base_id":"kb-foreign","knowledge_id":"doc-foreign","chunk_id":"chunk-foreign","source_id":"chunk-foreign","content":"外部正文篡改","embedding":[0.1,0.2]}` - chunkDoc := `{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-src","chunk_id":"chunk-src","source_id":"chunk-src","content":"来源正文一 resource://` + wsHandleSource + `","embedding":[0.5,1.5]}` - chunkDocTampered := `{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-src","chunk_id":"chunk-src","source_id":"chunk-src","content":"来源正文一 resource://` + wsHandleSource + `","embedding":[9.5,9.5]}` + mappedDoc := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src") + mappedTag := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src") + mappedChunk := expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src") + mappedIndex := expectedIndexID("chunk-src") + chunkJSON := func(embedding string) string { + return fmt.Sprintf(`{"tenant_id":%d,"knowledge_base_id":%q,"knowledge_id":%q,"chunk_id":%q,"source_id":%q,"content":%q,"embedding":%s}`, + wsTargetTenant, wsTargetKB, mappedDoc, mappedChunk, mappedChunk, content, embedding) + } + chunkDoc, chunkDocTampered := chunkJSON("[0.5,1.5]"), chunkJSON("[9.5,9.5]") + var originalChunk string + if err := fx.pool.QueryRow(fx.ctx, `SELECT content FROM chunks WHERE id=$1`, mappedChunk).Scan(&originalChunk); err != nil { + t.Fatalf("read published chunk content: %v", err) + } + execOne := func(sql string, args ...any) { + tag, err := fx.pool.Exec(fx.ctx, sql, args...) + if err != nil { + t.Fatalf("mutation sql: %v", err) + } + if tag.RowsAffected() != 1 { + t.Fatalf("mutation sql affected %d rows, want 1", tag.RowsAffected()) + } + } mutations := []struct { name string marker string @@ -1291,27 +1331,25 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) { restore func() }{ {"content", "chunks:", func() { - r10Exec(t, fx, `UPDATE chunks SET content=content||'!MUT' WHERE id='chunk-src'`) + execOne(`UPDATE chunks SET content=content||'!MUT' WHERE id=$1`, mappedChunk) }, func() { - if _, err := fx.pool.Exec(fx.ctx, `UPDATE chunks SET content=$1 WHERE id='chunk-src'`, content); err != nil { - t.Fatal(err) - } + execOne(`UPDATE chunks SET content=$1 WHERE id=$2`, originalChunk, mappedChunk) }}, - {"vector", "index doc chunk-src.embedding", func() { - putDoc("chunk-src", chunkDocTampered) + {"vector", "index doc " + expectedIndexID("chunk-src") + ".embedding", func() { + putDoc(mappedIndex, chunkDocTampered) }, func() { - putDoc("chunk-src", chunkDoc) + putDoc(mappedIndex, chunkDoc) }}, {"reference", "reference relation", func() { - r10Exec(t, fx, `UPDATE knowledge_tag_relations SET tag_id='missing-tag' WHERE knowledge_id='doc-src' AND tag_id='tag-src'`) + execOne(`UPDATE knowledge_tag_relations SET tag_id='missing-tag' WHERE knowledge_id=$1 AND tag_id=$2`, mappedDoc, mappedTag) }, func() { - r10Exec(t, fx, `UPDATE knowledge_tag_relations SET tag_id='tag-src' WHERE knowledge_id='doc-src' AND tag_id='missing-tag'`) + execOne(`UPDATE knowledge_tag_relations SET tag_id=$1 WHERE knowledge_id=$2 AND tag_id='missing-tag'`, mappedTag, mappedDoc) }}, // 这是日志内容变异，不是本地号映射变异：映射 RED 只在账本语义副本上做。 {"log-content", "wiki log", func() { - r10Exec(t, fx, `UPDATE wiki_log_entries SET action='tampered' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id='doc-src'`) + execOne(`UPDATE wiki_log_entries SET action='tampered' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id=$1`, mappedDoc) }, func() { - r10Exec(t, fx, `UPDATE wiki_log_entries SET action='ingest' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id='doc-src'`) + execOne(`UPDATE wiki_log_entries SET action='ingest' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id=$1`, mappedDoc) }}, {"foreign-index-source", "foreign index doc _source changed", func() { putDoc("chunk-foreign", foreignDocTampered) diff --git a/internal/releaseadapter/v071/library_chunk_sequence_integration_test.go b/internal/releaseadapter/v071/library_chunk_sequence_integration_test.go index 145bd35..0f21887 100644 --- a/internal/releaseadapter/v071/library_chunk_sequence_integration_test.go +++ b/internal/releaseadapter/v071/library_chunk_sequence_integration_test.go @@ -25,7 +25,7 @@ func TestLibraryChunkSequenceIsolated(t *testing.T) { t.Fatalf("cross-database publish must allocate local sequence: %+v", outcome) } var allocated int64 - if err := fx.pool.QueryRow(fx.ctx, `SELECT seq_id FROM chunks WHERE id='chunk-src'`).Scan(&allocated); err != nil { + if err := fx.pool.QueryRow(fx.ctx, `SELECT seq_id FROM chunks WHERE id=$1`, expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")).Scan(&allocated); err != nil { t.Fatal(err) } if allocated == occupied || allocated <= 0 { diff --git a/internal/releaseadapter/v071/library_index_recovery_gaps_test.go b/internal/releaseadapter/v071/library_index_recovery_gaps_test.go index 8988f26..c23545f 100644 --- a/internal/releaseadapter/v071/library_index_recovery_gaps_test.go +++ b/internal/releaseadapter/v071/library_index_recovery_gaps_test.go @@ -577,7 +577,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi var base wireCounts target.restoreMarkerFault = func(c libraryrelease.IndexWrite) error { - if c.ID != "chunk-src" { + if c.ID != expectedIndexID("chunk-src") { return nil } base = wire.counts() @@ -588,7 +588,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi t.Fatalf("state = %q (rollback=%q error=%q), want rollback_failed", outcome.State, outcome.Rollback, outcome.Error) } plan := outcome.Backup.WriteSet - change := indexChangeOf(t, plan, "chunk-src") + change := indexChangeOf(t, plan, expectedIndexID("chunk-src")) if len(change.Before) != 0 || len(change.After) == 0 { t.Fatalf("fixture change is not a created document: %+v", change) } @@ -597,7 +597,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi if refused := wire.since(base); refused.deletes != 0 || refused.clean != 0 || refused.creates != 0 { t.Fatalf("boundary fault dispatched %+v; refusal must send nothing", refused) } - if _, ok := recoveryIndexReadback(t, fx)["chunk-src"]; !ok { + if _, ok := recoveryIndexReadback(t, fx)[expectedIndexID("chunk-src")]; !ok { t.Fatal("precondition: the created document must still be present after the refused delete") } diff --git a/internal/releaseadapter/v071/library_production_target_integration_test.go b/internal/releaseadapter/v071/library_production_target_integration_test.go index 271bc2e..cffec5c 100644 --- a/internal/releaseadapter/v071/library_production_target_integration_test.go +++ b/internal/releaseadapter/v071/library_production_target_integration_test.go @@ -186,17 +186,21 @@ func TestLibraryProductionTargetIsolated(t *testing.T) { if err := target.Restore(ctx, secondBackup); err != nil { t.Fatalf("restore: %v", err) } - if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err != nil { - t.Fatalf("verify after restore: %v", err) + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: secondBackup.SnapshotID}); err != nil { + t.Fatalf("verify after restore to second job before-image: %v", err) } // Freeze repair evidence before fault injection, never from damaged state. backup := prepare(first) // 故障注入：直接破坏生产状态，校验必须失败，恢复后必须通过。 - if _, err := pool.Exec(ctx, `DELETE FROM chunks WHERE id='chunk-safe'`); err != nil { + deleted, err := pool.Exec(ctx, `DELETE FROM chunks WHERE id=$1`, expectedTargetID(10000, "kb-safe", "chunks", "chunk-safe")) + if err != nil { t.Fatal(err) } + if deleted.RowsAffected() != 1 { + t.Fatalf("fault injection deleted %d rows, want 1", deleted.RowsAffected()) + } if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err == nil { t.Fatal("verify must detect a missing chunk row") } @@ -650,14 +654,15 @@ func TestLibraryCrossScopeSyncRollbackLeavesNoResidueIsolated(t *testing.T) { } var knowledgeKB, chunkKB, provider, locator string var tenant int64 - if err := pool.QueryRow(ctx, `SELECT tenant_id, knowledge_base_id FROM knowledges WHERE id='doc-source'`). + if err := pool.QueryRow(ctx, `SELECT tenant_id, knowledge_base_id FROM knowledges WHERE id=$1`, + expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-source")). Scan(&tenant, &knowledgeKB); err != nil { t.Fatalf("source document must exist in the target scope: %v", err) } if tenant != 10001 || knowledgeKB != "kb-target" { t.Fatalf("cross-scope row landed at %d/%s", tenant, knowledgeKB) } - if err := pool.QueryRow(ctx, `SELECT knowledge_base_id FROM chunks WHERE id='chunk-source'`).Scan(&chunkKB); err != nil { + if err := pool.QueryRow(ctx, `SELECT knowledge_base_id FROM chunks WHERE id=$1`, expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-source")).Scan(&chunkKB); err != nil { t.Fatal(err) } if chunkKB != "kb-target" { diff --git a/internal/releaseadapter/v071/library_target_ids.go b/internal/releaseadapter/v071/library_target_ids.go index ac300d2..ee05e6e 100644 --- a/internal/releaseadapter/v071/library_target_ids.go +++ b/internal/releaseadapter/v071/library_target_ids.go @@ -28,14 +28,13 @@ import ( // 资源表（resources）是多知识库共享的字节载体，ID 不重映射，沿用既有复用/归属证明； // 绑定行的 owner_id 与自身 id 则按目标映射。 type targetIDMapper struct { - tenant uint64 - kb string - ids map[string]string - reverse map[string]string - known map[string]map[string]bool - owner map[string]string - ambiguous map[string]bool - sequence int + tenant uint64 + kb string + ids map[string]string + reverse map[string]string + known map[string]map[string]bool + owner map[string]string + sequence int // 不重映射的表：资源行是跨库共享对象，改了 ID 就等于复制一份字节， // 现有复用契约要求"内容相同 + 来源可审计"才复用，不能靠换 ID 绕过。 @@ -50,7 +49,7 @@ func newTargetIDMapper(tenant uint64, kb string, _ any) *targetIDMapper { return &targetIDMapper{ tenant: tenant, kb: kb, ids: map[string]string{}, reverse: map[string]string{}, - known: map[string]map[string]bool{}, owner: map[string]string{}, ambiguous: map[string]bool{}, + known: map[string]map[string]bool{}, owner: map[string]string{}, skip: map[string]bool{"resources": true}, } } @@ -85,19 +84,17 @@ func (m *targetIDMapper) registerSourceID(table, id string) { } // 本地自增主键（如 wiki_log_entries.id）不是跨库身份：既不能当引用匹配， // 也不能参与"残留源 ID"判断，否则普通整数列会被误报。 - if contract, ok := migrationIdentityContracts[table]; ok && contract.Kind == "local-bigint" { + if contract, ok := migrationIdentityContracts[table]; ok && (contract.Kind == "local-bigint" || contract.Kind == "tuple") { + // 本地自增/复合元组都不是可被别的表引用的实体身份： + // 关系表的两个分量必须由 knowledges/knowledge_tags 自己登记。 return } if m.known[table] == nil { m.known[table] = map[string]bool{} } m.known[table][id] = true - if previous, exists := m.owner[id]; !exists { + if _, exists := m.owner[id]; !exists { m.owner[id] = table - } else if previous != table { - // 两张表出现同一字符串 ID：无法从值本身判定引用指向哪张表， - // 只能报歧义，不能 first-table-wins 猜一个。 - m.ambiguous[id] = true } } @@ -127,7 +124,9 @@ func (m *targetIDMapper) remapRow(table string, row map[string]json.RawMessage) } } for _, column := range []string{"relation_chunks", "indirect_relation_chunks"} { - if err := m.remapJSONExact(row, column); err != nil { + // 这两列是"分块引用"：目标表由字段语义决定（chunks），不由字符串 + // 在别的表里也出现来决定；跨表同 ID 是合法数据，不是歧义。 + if err := m.remapJSONExact(row, column, "chunks"); err != nil { return err } } @@ -166,6 +165,9 @@ func (m *targetIDMapper) remapRow(table string, row map[string]json.RawMessage) if err := m.remapScalar(row, "id", table); err != nil { return err } + if err := m.remapRefArray(row, "suspected_knowledge_ids", "knowledges"); err != nil { + return err + } for _, column := range []string{"page_id", "knowledge_id", "chunk_id"} { if err := m.remapScalar(row, column, m.tableFor(column)); err != nil { return err @@ -294,7 +296,7 @@ func (m *targetIDMapper) remapSourceRefs(row map[string]json.RawMessage) error { // remapJSONExact 只在指定 JSON 列里替换"整体等于已知身份"的字符串； // 不做子串替换（正文里出现同形状文本不是引用）。 -func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column string) error { +func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column, table string) error { raw, ok := row[column] if !ok { return nil @@ -303,35 +305,51 @@ func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column s decoder.UseNumber() // >2^53 的整数必须原样保留，不能经 float64 var value any if err := decoder.Decode(&value); err != nil { - return fmt.Errorf("chunks.%s 不可解析", column) + return fmt.Errorf("%s.%s 不可解析", table, column) } changed := false - var walk func(any) any - walk = func(v any) any { + var walk func(any) (any, error) + walk = func(v any) (any, error) { switch typed := v.(type) { case string: - if m.ambiguous[typed] { - return typed // 由调用方在扫描阶段报歧义 - } - if table, ok := m.owner[typed]; ok && !m.skip[table] { + if m.known[table][typed] { changed = true - return m.mapID(table, typed) + return m.mapID(table, typed), nil } - return typed + for other, ids := range m.known { + if other == table || m.skip[other] { + continue + } + if ids[typed] { + return nil, fmt.Errorf("%w: %s.%s references %s entity %s", errUnmappedTargetReference, table, column, other, typed) + } + } + return typed, nil case []any: for i := range typed { - typed[i] = walk(typed[i]) + mapped, err := walk(typed[i]) + if err != nil { + return nil, err + } + typed[i] = mapped } - return typed + return typed, nil case map[string]any: for key := range typed { - typed[key] = walk(typed[key]) + mapped, err := walk(typed[key]) + if err != nil { + return nil, err + } + typed[key] = mapped } - return typed + return typed, nil } - return v + return v, nil + } + mapped, err := walk(value) + if err != nil { + return err } - mapped := walk(value) if !changed { return nil } @@ -345,11 +363,6 @@ func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column s // rejectUnmapped 保证"改写后不再残留任何已知源身份"，并拒绝无法判定归属的悬挂引用。 func (m *targetIDMapper) rejectUnmapped(table string, row map[string]json.RawMessage) error { - for id := range m.ambiguous { - if strings.Contains(strings.Join(rawStrings(row), "\x00"), id) { - return fmt.Errorf("%w: ambiguous id %s appears in %s", errUnmappedTargetReference, id, table) - } - } body := map[string]bool{"content": true, "summary": true, "description": true, "physical_path": true, "pages_affected": true, "slug": true, "metadata": true, "page_metadata": true, "title": true, "aliases": true, "in_links": true, "out_links": true} for column, raw := range row { @@ -409,6 +422,21 @@ func (m *targetIDMapper) remapIndexDocument(id string, source json.RawMessage) ( encoded, _ := json.Marshal(m.mapID(table, value)) fields[column] = encoded } + // source_id 在不同索引器版本里既可能是 chunk id，也可能是外部文档 id： + // 只有能确定是分块/文档身份时才映射，opaque 外部 id 原样保留。 + if raw, ok := fields["source_id"]; ok { + var value string + if json.Unmarshal(raw, &value) == nil && value != "" { + switch { + case m.known["chunks"][value]: + encoded, _ := json.Marshal(m.mapID("chunks", value)) + fields["source_id"] = encoded + case m.known["knowledges"][value]: + encoded, _ := json.Marshal(m.mapID("knowledges", value)) + fields["source_id"] = encoded + } + } + } encoded, err := json.Marshal(fields) if err != nil { return "", nil, err diff --git a/internal/releaseadapter/v071/library_target_ids_integration_test.go b/internal/releaseadapter/v071/library_target_ids_integration_test.go index c13de2a..ab7d40a 100644 --- a/internal/releaseadapter/v071/library_target_ids_integration_test.go +++ b/internal/releaseadapter/v071/library_target_ids_integration_test.go @@ -2,8 +2,11 @@ package v071 import ( "context" + "strings" "testing" + "knowledge-governance/internal/librarysnapshot" + "knowledge-governance/internal/libraryrelease" ) @@ -113,3 +116,56 @@ func TestTargetIDMappingPlanApplyVerifyRestoreIsolated(t *testing.T) { } t.Logf("plan rows=%d index=%d objects=%d; old kb preserved; restore exact", len(set.Rows), len(set.Index), len(set.Objects)) } + +// 映射后的目标 ID 若真被另一个库占用，仍必须拒绝（不能只取消旧冲突预期）。 +func TestTargetIDMappedIdentityCollisionStillRejectedIsolated(t *testing.T) { + fx := openWriteSetFixture(t) + seedTargetData(t, fx) + mapped := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src") + if _, err := fx.pool.Exec(fx.ctx, `INSERT INTO knowledges VALUES($1,10001,$2,'completed',0,'',NULL)`, mapped, wsOtherKB); err != nil { + t.Fatal(err) + } + snapshot := sealSourceSnapshot(t, fx, nil) + target := fx.target(t) + current, err := target.Current(fx.ctx) + if err != nil { + t.Fatal(err) + } + backup, err := fx.backups(t, target).Prepare(fx.ctx, snapshot, current) + if err != nil { + t.Fatal(err) + } + if _, err := target.PlanWriteSet(fx.ctx, libraryrelease.WriteSetPlan{JobID: "mapped-collision", Target: "internal", + BackupSnapshotID: backup.SnapshotID, TargetSnapshotID: snapshot, Universe: backup.Universe}); err == nil { + t.Fatal("mapped identity occupied by another KB was accepted") + } +} + +// 旧空 mode 计划没有映射契约：真实写入路径必须按原 ID 落库并因此与旧库冲突， +// 证明新进程不会给历史计划隐式重映射。 +func TestLegacyPlanWithoutModeDoesNotRemapIsolated(t *testing.T) { + fx := openWriteSetFixture(t) + seedTargetData(t, fx) + if _, err := fx.pool.Exec(fx.ctx, `INSERT INTO knowledges VALUES('doc-src',10001,$1,'completed',0,'',NULL)`, wsOtherKB); err != nil { + t.Fatal(err) + } + snapshot := sealSourceSnapshot(t, fx, nil) + target := fx.target(t) + archive, err := librarysnapshot.Open(fx.archiveRoot, 256*1024*1024) + if err != nil { + t.Fatal(err) + } + defer archive.Close() + manifest, err := archive.Load(snapshot) + if err != nil { + t.Fatal(err) + } + target.disableTargetMapping() + if err := target.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil { + t.Fatal(err) + } + err = target.importRows(context.Background(), archive, manifest, "database") + if err == nil || !strings.Contains(strings.ToLower(err.Error()), "duplicate") && !strings.Contains(err.Error(), "冲突") { + t.Fatalf("legacy path must write original ids and fail on the old-kb collision, got %v", err) + } +} diff --git a/internal/releaseadapter/v071/library_target_ids_test.go b/internal/releaseadapter/v071/library_target_ids_test.go index 61cfcea..1359086 100644 --- a/internal/releaseadapter/v071/library_target_ids_test.go +++ b/internal/releaseadapter/v071/library_target_ids_test.go @@ -195,7 +195,7 @@ func TestTargetIDMapperRewritesRelationTuple(t *testing.T) { func TestTargetMappingOnlyAppliesToTargetSnapshot(t *testing.T) { plan := &libraryrelease.WriteSet{TargetSnapshotID: "target-seal", BackupSnapshotID: "backup-seal"} - target := &ProductionLibraryTarget{activePlan: plan} + target := &ProductionLibraryTarget{activePlan: plan, targetIDMode: libraryrelease.TargetIDModeScoped} if !target.targetMappingApplies("target-seal") { t.Fatal("target snapshot must be mapped") } @@ -206,6 +206,11 @@ func TestTargetMappingOnlyAppliesToTargetSnapshot(t *testing.T) { if target.targetMappingApplies("target-seal") { t.Fatal("restore replay must not be mapped") } + target.preserveChunkSequence = false + target.targetIDMode = "" + if target.targetMappingApplies("target-seal") { + t.Fatal("legacy plan without target id mode must not be mapped") + } } func TestTargetIDMapperPreservesLargeNumbersAndRejectsAmbiguity(t *testing.T) { @@ -226,10 +231,14 @@ func TestTargetIDMapperPreservesLargeNumbersAndRejectsAmbiguity(t *testing.T) { same := "17171717-1717-4717-8717-171717171717" m.registerSourceID("chunks", same) m.registerSourceID("knowledges", same) - bad := rawTargetRow(t, map[string]any{"id": "18181818-1818-4818-8818-181818181818", + m.registerSourceID("chunks", "18181818-1818-4818-8818-181818181818") + shared := rawTargetRow(t, map[string]any{"id": "18181818-1818-4818-8818-181818181818", "relation_chunks": []string{same}}) - if err := m.remapRow("chunks", bad); err == nil { - t.Fatal("ambiguous cross-table id was mapped by guess") + if err := m.remapRow("chunks", shared); err != nil { + t.Fatalf("typed chunk reference rejected for cross-table same id: %v", err) + } + if got := string(shared["relation_chunks"]); got != `["`+m.mapID("chunks", same)+`"]` { + t.Fatalf("relation must resolve by field semantics: %s", got) } } diff --git a/internal/releaseadapter/v071/library_target_path.go b/internal/releaseadapter/v071/library_target_path.go index 5d92929..34eb995 100644 --- a/internal/releaseadapter/v071/library_target_path.go +++ b/internal/releaseadapter/v071/library_target_path.go @@ -6,6 +6,8 @@ import ( "fmt" "sort" + "knowledge-governance/internal/libraryrelease" + "knowledge-governance/internal/librarysnapshot" ) @@ -32,13 +34,15 @@ func (t *ProductionLibraryTarget) disableTargetMapping() { // targetMappingApplies 只在"把目标快照写进目标"时启用映射； // 备份/恢复快照里的 ID 是目标自己的既有身份，二次映射会把恢复写坏。 func (t *ProductionLibraryTarget) targetMappingApplies(snapshotID string) bool { - if t.preserveChunkSequence || snapshotID == "" { + // 只有持久计划显式声明 scoped 契约时才映射。空 mode 是历史计划： + // 它登记的就是原身份，重映射会把恢复/重试写到别的 ID 上。 + if t.targetIDMode != libraryrelease.TargetIDModeScoped || t.preserveChunkSequence || snapshotID == "" { return false } - if t.activePlan != nil { - return snapshotID == t.activePlan.TargetSnapshotID + if t.activePlan == nil { + return false } - return true + return snapshotID == t.activePlan.TargetSnapshotID } // archivedRowIdentitiesFor 返回写入后的目标身份集合（映射开启时）或归档原身份。 diff --git a/internal/releaseadapter/v071/library_verify_content.go b/internal/releaseadapter/v071/library_verify_content.go index d34185e..a5c4eda 100644 --- a/internal/releaseadapter/v071/library_verify_content.go +++ b/internal/releaseadapter/v071/library_verify_content.go @@ -261,6 +261,9 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive } archived = append(archived, SearchDocument{ID: document.ID, Source: canonical}) } + // 目标 ID 映射会改变文档 ID 的排序位置；摘要按 ID 排序后的集合计算， + // 否则少一个/多一个的差异会被误报成"数量一致但摘要不同"。 + sort.Slice(archived, func(i, j int) bool { return archived[i].ID < archived[j].ID }) wantDigest, err := searchDocumentsDigest(archived) if err != nil { return err diff --git a/internal/releaseadapter/v071/target_id_expect_test.go b/internal/releaseadapter/v071/target_id_expect_test.go new file mode 100644 index 0000000..958cb96 --- /dev/null +++ b/internal/releaseadapter/v071/target_id_expect_test.go @@ -0,0 +1,131 @@ +package v071 + +import ( + "crypto/sha256" + "fmt" + "strings" + "testing" +) + +// 测试侧独立实现的目标 ID 推导：公式来自产品文档（scoped-sha256-v1）， +// 不复用生产 mapper/rowTransform，避免"用实现证明实现"。 +func expectedTargetID(tenant uint64, kb, table, source string) string { + sum := sha256.Sum256([]byte("knowledge-governance/target-identity/v1\x00" + + fmt.Sprintf("%d", tenant) + "\x00" + kb + "\x00" + table + "\x00" + source)) + var raw [16]byte + copy(raw[:], sum[:16]) + raw[6] = raw[6]&0x0f | 0x50 + raw[8] = raw[8]&0x3f | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", raw[0:4], raw[4:6], raw[6:8], raw[8:10], raw[10:16]) +} + +func expectedIndexID(source string) string { + return expectedTargetID(wsTargetTenant, wsTargetKB, "index_chunks", source) +} + +// expectedTargetRow 按已知引用字段语义改写一行期望数据，供独立 oracle 使用。 +func expectedTargetRow(table string, row map[string]any) map[string]any { + out := make(map[string]any, len(row)) + for k, v := range row { + out[k] = v + } + str := func(key string) (string, bool) { s, ok := out[key].(string); return s, ok } + mapped := func(key, targetTable string) { + if s, ok := str(key); ok && s != "" { + out[key] = expectedTargetID(wsTargetTenant, wsTargetKB, targetTable, s) + } + } + switch table { + case "knowledges": + mapped("id", "knowledges") + case "chunks": + mapped("id", "chunks") + mapped("knowledge_id", "knowledges") + mapped("tag_id", "knowledge_tags") + for _, key := range []string{"parent_chunk_id", "pre_chunk_id", "next_chunk_id"} { + mapped(key, "chunks") + } + for _, key := range []string{"relation_chunks", "indirect_relation_chunks"} { + if list, ok := out[key].([]any); ok { + for i, item := range list { + if s, ok := item.(string); ok { + list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", s) + } + } + } + } + case "knowledge_tags": + mapped("id", "knowledge_tags") + case "knowledge_tag_relations": + mapped("knowledge_id", "knowledges") + mapped("tag_id", "knowledge_tags") + case "wiki_folders": + mapped("id", "wiki_folders") + mapped("parent_id", "wiki_folders") + case "wiki_pages": + mapped("id", "wiki_pages") + mapped("folder_id", "wiki_folders") + if list, ok := out["chunk_refs"].([]any); ok { + for i, item := range list { + if s, ok := item.(string); ok { + list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", s) + } + } + } + if list, ok := out["source_refs"].([]any); ok { + for i, item := range list { + if s, ok := item.(string); ok { + id, suffix := s, "" + if cut := strings.IndexByte(s, '|'); cut >= 0 { + id, suffix = s[:cut], s[cut:] + } + list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", id) + suffix + } + } + } + case "wiki_page_issues": + mapped("id", "wiki_page_issues") + mapped("knowledge_id", "knowledges") + mapped("page_id", "wiki_pages") + if list, ok := out["suspected_knowledge_ids"].([]any); ok { + for i, item := range list { + if s, ok := item.(string); ok { + list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", s) + } + } + } + case "wiki_log_entries": + mapped("knowledge_id", "knowledges") + mapped("page_id", "wiki_pages") + case "resource_bindings": + mapped("id", "resource_bindings") + if ownerType, _ := out["owner_type"].(string); ownerType == "knowledge" { + mapped("owner_id", "knowledges") + } else if ownerType == "chunk" { + mapped("owner_id", "chunks") + } + } + return out +} + +// expectedIndexSource 按语义改写索引文档里的引用（正文/向量不动）。 +func expectedIndexSource(t *testing.T, source map[string]any) map[string]any { + t.Helper() + out := make(map[string]any, len(source)) + for k, v := range source { + out[k] = v + } + for key, targetTable := range map[string]string{"chunk_id": "chunks", "knowledge_id": "knowledges", "tag_id": "knowledge_tags"} { + if s, ok := out[key].(string); ok && s != "" { + out[key] = expectedTargetID(wsTargetTenant, wsTargetKB, targetTable, s) + } + } + if s, ok := out["tenant_id"]; ok { + if number, ok := s.(float64); ok && number == float64(wsSourceTenant) { + out["tenant_id"] = float64(wsTargetTenant) + } + } + out["tenant_id"] = float64(wsTargetTenant) + out["knowledge_base_id"] = wsTargetKB + return out +}