package legacygraph import ( "context" "crypto/sha256" "encoding/binary" "encoding/hex" "encoding/json" "fmt" "slices" "sort" "strings" "unicode" "github.com/Tencent/WeKnora/internal/types" "gorm.io/gorm" ) const nativeOrigin = "native_extraction_v1" type sourceChunk struct { ID, KnowledgeID, KnowledgeBaseID, Content, FileName string } func contentHash(content string) string { h := sha256.Sum256([]byte(content)) return hex.EncodeToString(h[:]) } // IDs stay exactly representable in the browser, outside historical IDs. The // full identity is also checked on conflict, so a hash collision fails closed. func nativeID(parts ...string) (int64, string) { identity, _ := json.Marshal(parts) h := sha256.Sum256(identity) return int64(binary.BigEndian.Uint64(h[:8])&((1<<52)-1)) | 1<<52, hex.EncodeToString(h[:]) } func normalizedName(name string) string { return strings.ToLower(strings.TrimSpace(name)) } func sourceScopedEntity(name string) bool { name = strings.Map(func(r rune) rune { if unicode.IsSpace(r) { return -1 } return r }, normalizedName(name)) switch name { case "本品", "该产品", "本产品", "产品", "本商品", "此产品", "本公司", "该公司", "本文", "本方法", "该方法", "thisproduct", "theproduct": return true default: // The extractor uses these names to separate unresolved local products. return strings.HasSuffix(name, "未名产品") || strings.HasPrefix(name, "本品(") } } func nativeEntityIdentity(document string, node *types.GraphNode) (int64, string, error) { key := normalizedName(node.Name) parts := []string{document, "entity", key} if sourceScopedEntity(key) { // An unnamed referent is not a document-wide entity. Multiple sources // require the extractor to supply an unambiguous, specifically named node. if len(node.Chunks) == 0 || node.Chunks[0] == "" { return 0, "", fmt.Errorf("generic graph entity requires one source chunk") } for _, chunk := range node.Chunks { if chunk != node.Chunks[0] { return 0, "", fmt.Errorf("generic graph entity cannot merge distinct source chunks") } } parts = append(parts, "source", node.Chunks[0]) } id, identity := nativeID(parts...) return id, identity, nil } // HasChunkGraph accepts only current checkpoints with extracted relations. // Legacy empty results and excerpt mappings require coverage review. func (r *Repository) HasChunkGraph(ctx context.Context, chunk *types.Chunk) (bool, error) { tenant, ok := types.TenantIDFromContext(ctx) if !ok || tenant == 0 || chunk == nil || chunk.TenantID != tenant || r.db == nil { return false, fmt.Errorf("graph checkpoint requires source tenant and chunk") } var count int64 err := r.db.WithContext(ctx).Table("native_graph_records").Where("tenant_id=? AND knowledge_base_id=? AND source_table='knowledgegraphchunkstate' AND payload->>'origin'=? AND payload->>'native_chunk_id'=? AND payload->>'native_knowledge_id'=? AND payload->>'source_content_hash'=? AND payload->>'extraction_status'='complete' AND CAST(COALESCE(payload->>'relation_count','0') AS INTEGER)>0", tenant, chunk.KnowledgeBaseID, nativeOrigin, chunk.ID, chunk.KnowledgeID, contentHash(chunk.Content)).Count(&count).Error return count > 0, err } // AddChunkGraph rejects unreviewed empty coverage and results from changed sources. func (r *Repository) AddChunkGraph(ctx context.Context, ns types.NameSpace, chunk *types.Chunk, graph *types.GraphData) error { if chunk == nil || chunk.ID == "" || graph == nil { return fmt.Errorf("graph extraction requires its source chunk and model result") } if len(graph.Relation) == 0 { return fmt.Errorf("graph extraction without relations requires coverage review") } return r.addNativeGraphs(ctx, ns, []*types.GraphData{graph}, map[string]string{chunk.ID: contentHash(chunk.Content)}) } func (r *Repository) addNativeGraphs(ctx context.Context, ns types.NameSpace, graphs []*types.GraphData, expected map[string]string) error { tenant, ok := types.TenantIDFromContext(ctx) if !ok || tenant == 0 || ns.KnowledgeBase == "" || ns.Knowledge == "" || r.db == nil { return fmt.Errorf("graph writes require tenant, knowledge base, document and database") } ids := map[string]bool{} for id := range expected { ids[id] = true } for _, graph := range graphs { if graph == nil { return fmt.Errorf("nil graph extraction") } for _, node := range graph.Node { if node == nil || normalizedName(node.Name) == "" || len(node.Chunks) == 0 { return fmt.Errorf("extracted graph node requires name and source chunks") } for _, id := range node.Chunks { if id == "" || (expected != nil && expected[id] == "") { return fmt.Errorf("graph node refers to a chunk outside extraction scope") } ids[id] = true } } } if len(ids) == 0 { return fmt.Errorf("graph extraction has no source chunks") } chunkIDs := make([]string, 0, len(ids)) for id := range ids { chunkIDs = append(chunkIDs, id) } sort.Strings(chunkIDs) return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if err := lockGraph(tx, tenant, ns.KnowledgeBase); err != nil { return err } var chunks []sourceChunk if err := tx.Raw(`SELECT c.id,c.knowledge_id,c.knowledge_base_id,c.content,k.file_name FROM chunks c JOIN knowledges k ON k.id=c.knowledge_id AND k.tenant_id=c.tenant_id AND k.knowledge_base_id=c.knowledge_base_id AND k.deleted_at IS NULL JOIN knowledge_bases kb ON kb.id=c.knowledge_base_id AND kb.tenant_id=c.tenant_id AND kb.deleted_at IS NULL WHERE c.tenant_id=? AND c.knowledge_base_id=? AND c.knowledge_id=? AND c.id IN ? AND c.deleted_at IS NULL AND c.is_enabled=true ORDER BY c.id FOR SHARE OF c,k,kb`, tenant, ns.KnowledgeBase, ns.Knowledge, chunkIDs).Scan(&chunks).Error; err != nil { return err } if len(chunks) != len(chunkIDs) { return fmt.Errorf("graph source chunks are unavailable or outside document/tenant scope") } byID := map[string]sourceChunk{} for _, chunk := range chunks { if expected != nil && expected[chunk.ID] != contentHash(chunk.Content) { return fmt.Errorf("graph source chunk %s changed during extraction", chunk.ID) } byID[chunk.ID] = chunk } // Replace only this extraction's evidence. Other documents, chunks and // archived migration records are untouched, including on retries. sources := make([]string, 0, len(chunks)) for _, chunk := range chunks { source := "native:" + chunk.ID sources = append(sources, source) excerpt := []rune(chunk.Content) if len(excerpt) > 2000 { excerpt = excerpt[:2000] } if err := tx.Exec(`INSERT INTO native_graph_sources(tenant_id,knowledge_base_id,source_chunk_id,source_file,source_content_hash,excerpt,native_knowledge_id,native_chunk_id,native_content_sha256,mapping_state) VALUES(?,?,?,?,?,?,?,?,?,'native_extraction') ON CONFLICT(tenant_id,knowledge_base_id,source_chunk_id) DO UPDATE SET source_file=excluded.source_file,source_content_hash=excluded.source_content_hash,excerpt=excluded.excerpt,native_knowledge_id=excluded.native_knowledge_id,native_chunk_id=excluded.native_chunk_id,native_content_sha256=excluded.native_content_sha256,mapping_state=excluded.mapping_state`, tenant, ns.KnowledgeBase, source, chunk.FileName, contentHash(chunk.Content), string(excerpt), ns.Knowledge, chunk.ID, contentHash(chunk.Content)).Error; err != nil { return err } } type pendingEntity struct { identity string payload map[string]any } entities := map[int64]pendingEntity{} // Removed nodes also lose their source's attributes, even when a bare // reference from another chunk keeps the entity alive. var replacedEntities []Record if err := tx.Raw(`SELECT r.legacy_id,r.payload FROM native_graph_records r WHERE r.tenant_id=? AND r.knowledge_base_id=? AND r.source_table='knowledgeentity' AND r.payload->>'origin'=? AND r.payload->>'native_knowledge_id'=? AND EXISTS (SELECT 1 FROM native_graph_records e WHERE e.tenant_id=r.tenant_id AND e.knowledge_base_id=r.knowledge_base_id AND e.source_table='knowledgegraphevidence' AND e.payload->>'origin'=? AND e.payload->>'subject_type'='entity' AND e.payload->>'subject_id'=r.legacy_id::text AND e.payload->>'source_chunk_id' IN ?)`, tenant, ns.KnowledgeBase, nativeOrigin, ns.Knowledge, nativeOrigin, sources).Scan(&replacedEntities).Error; err != nil { return err } for _, record := range replacedEntities { var payload map[string]any if err := json.Unmarshal(record.Payload, &payload); err != nil { return err } identity, _ := payload["native_identity"].(string) entities[record.ID] = pendingEntity{identity, payload} } if err := tx.Exec(`DELETE FROM native_graph_records WHERE tenant_id=? AND knowledge_base_id=? AND source_table='knowledgegraphevidence' AND payload->>'origin'=? AND payload->>'source_chunk_id' IN ?`, tenant, ns.KnowledgeBase, nativeOrigin, sources).Error; err != nil { return err } counts := map[string]map[string]int{} entitySources := map[int64]map[string][]string{} for _, id := range chunkIDs { counts[id] = map[string]int{"entity": 0, "relation": 0} } for _, graph := range graphs { nodes := map[string]*types.GraphNode{} entityIDs := map[string]int64{} for _, node := range graph.Node { key := normalizedName(node.Name) id, identity, err := nativeEntityIdentity(ns.Knowledge, node) if err != nil { return err } if priorID, exists := entityIDs[key]; exists && priorID != id { return fmt.Errorf("generic graph entity is ambiguous within extraction") } name := strings.TrimSpace(node.Name) payload := map[string]any{"canonical_name": name, "entity_type": "concept", "attributes": node.Attributes} if len(node.Attributes) > 0 { payload["entity_type"] = node.Attributes[0] } if prior, exists := entities[id]; exists && prior.identity != identity { return fmt.Errorf("native graph identity collision; preserved graph was not modified") } if _, exists := entitySources[id]; !exists { entities[id] = pendingEntity{identity, payload} entitySources[id] = map[string][]string{} } else if name < entities[id].payload["canonical_name"].(string) { entities[id].payload["canonical_name"] = name } if prior := nodes[key]; prior != nil { prior.Chunks = append(prior.Chunks, node.Chunks...) } else { copyNode := *node copyNode.Chunks = append([]string(nil), node.Chunks...) nodes[key] = ©Node } entityIDs[key] = id seen := map[string]bool{} for _, chunkID := range node.Chunks { if seen[chunkID] { continue } seen[chunkID] = true if attributes, exists := entitySources[id][chunkID]; exists { if !slices.Equal(attributes, node.Attributes) { return fmt.Errorf("duplicate native graph entity source attributes conflict; extraction requires review") } continue } entitySources[id][chunkID] = node.Attributes if err := putEvidence(tx, tenant, ns, "entity", id, byID[chunkID], "", node.Attributes); err != nil { return err } counts[chunkID]["entity"]++ } } for _, rel := range graph.Relation { if rel == nil || strings.TrimSpace(rel.Type) == "" { return fmt.Errorf("invalid extracted relation") } from, to := normalizedName(rel.Node1), normalizedName(rel.Node2) if nodes[from] == nil || nodes[to] == nil { return fmt.Errorf("relation references an entity outside extraction") } shared := []string{} fromChunks := map[string]bool{} for _, id := range nodes[from].Chunks { fromChunks[id] = true } for _, id := range nodes[to].Chunks { if fromChunks[id] { shared = append(shared, id) delete(fromChunks, id) } } if len(shared) == 0 { return fmt.Errorf("relation has no shared source evidence") } if rel.Evidence != "" { grounded := shared[:0] for _, chunkID := range shared { if strings.Contains(byID[chunkID].Content, rel.Evidence) { grounded = append(grounded, chunkID) } } shared = grounded if len(shared) == 0 { return fmt.Errorf("relation quotation is absent from current source evidence") } } identityParts := []string{ns.Knowledge, "relation", from, to, normalizedName(rel.Type)} if sourceScopedEntity(from) || sourceScopedEntity(to) { identityParts = append(identityParts, "scoped_endpoints", fmt.Sprint(entityIDs[from]), fmt.Sprint(entityIDs[to])) } id, identity := nativeID(identityParts...) if err := putNativeRecord(tx, tenant, ns, "knowledgerelation", id, identity, map[string]any{"from_entity_id": entityIDs[from], "to_entity_id": entityIDs[to], "relation_type": strings.TrimSpace(rel.Type), "evidence": rel.Evidence}, ""); err != nil { return err } for _, chunkID := range shared { if err := putEvidence(tx, tenant, ns, "relation", id, byID[chunkID], rel.Evidence); err != nil { return err } counts[chunkID]["relation"]++ } } } // Reconcile after every source snapshot is present, so batch ordering // cannot decide which source owns the entity's retained attributes. entityIDs := make([]int64, 0, len(entities)) for id := range entities { entityIDs = append(entityIDs, id) } slices.Sort(entityIDs) for _, id := range entityIDs { entity := entities[id] if err := putNativeRecord(tx, tenant, ns, "knowledgeentity", id, entity.identity, entity.payload, ""); err != nil { return err } } for _, chunk := range chunks { id, identity := nativeID(ns.Knowledge, "chunkstate", chunk.ID) status := "complete" if counts[chunk.ID]["relation"] == 0 { status = "needs_review" } if err := putNativeRecord(tx, tenant, ns, "knowledgegraphchunkstate", id, identity, map[string]any{"source_chunk_id": "native:" + chunk.ID, "native_chunk_id": chunk.ID, "source_content_hash": contentHash(chunk.Content), "extraction_status": status, "entity_count": counts[chunk.ID]["entity"], "relation_count": counts[chunk.ID]["relation"]}, contentHash(chunk.Content)); err != nil { return err } } return pruneNativeOrphans(tx, tenant, ns) }) } func lockGraph(tx *gorm.DB, tenant uint64, kb string) error { return tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?,0))", fmt.Sprintf("native-graph:%d:%s", tenant, kb)).Error } func putNativeRecord(tx *gorm.DB, tenant uint64, ns types.NameSpace, table string, id int64, identity string, payload map[string]any, hash string) error { if table == "knowledgeentity" { if err := reconcileNativeEntityAttributes(tx, tenant, ns, id, identity, payload); err != nil { return err } } payload["origin"], payload["native_identity"], payload["native_knowledge_id"], payload["status"] = nativeOrigin, identity, ns.Knowledge, "active" data, err := json.Marshal(payload) if err != nil { return err } if hash == "" { hash = contentHash(string(data)) } result := tx.Exec(`INSERT INTO native_graph_records(tenant_id,knowledge_base_id,source_table,legacy_id,graph_version_id,payload,source_snapshot_sha256) VALUES(?,?,?,?,0,?::jsonb,?) ON CONFLICT(tenant_id,knowledge_base_id,source_table,legacy_id) DO UPDATE SET payload=excluded.payload,source_snapshot_sha256=excluded.source_snapshot_sha256 WHERE native_graph_records.payload->>'native_identity'=excluded.payload->>'native_identity' AND native_graph_records.payload->>'origin'=?`, tenant, ns.KnowledgeBase, table, id, string(data), hash, nativeOrigin) if result.Error != nil { return result.Error } if result.RowsAffected != 1 { return fmt.Errorf("native graph identity collision; preserved graph was not modified") } return nil } func reconcileNativeEntityAttributes(tx *gorm.DB, tenant uint64, ns types.NameSpace, id int64, identity string, payload map[string]any) error { var evidence []struct { Payload json.RawMessage } // Evidence for the replaced sources has already been rewritten with raw // node attributes. Only still-current sources may preserve shared facts. if err := tx.Raw(`SELECT e.payload FROM native_graph_records e JOIN native_graph_sources s ON s.tenant_id=e.tenant_id AND s.knowledge_base_id=e.knowledge_base_id AND s.source_chunk_id=e.payload->>'source_chunk_id' JOIN chunks c ON c.id=s.native_chunk_id AND c.tenant_id=s.tenant_id AND c.knowledge_base_id=s.knowledge_base_id AND c.deleted_at IS NULL AND c.is_enabled=true AND encode(sha256(convert_to(c.content,'UTF8')),'hex')=s.native_content_sha256 JOIN knowledges k ON k.id=c.knowledge_id AND k.tenant_id=c.tenant_id AND k.knowledge_base_id=c.knowledge_base_id AND k.deleted_at IS NULL JOIN knowledge_bases kb ON kb.id=c.knowledge_base_id AND kb.tenant_id=c.tenant_id AND kb.deleted_at IS NULL WHERE e.tenant_id=? AND e.knowledge_base_id=? AND e.source_table='knowledgegraphevidence' AND e.payload->>'subject_type'='entity' AND e.payload->>'subject_id'=? AND e.payload->>'origin'=? AND e.payload->>'status'='active' AND e.payload->>'native_knowledge_id'=? AND c.knowledge_id=? AND s.native_knowledge_id=c.knowledge_id AND s.mapping_state='native_extraction' AND e.payload->>'source_content_hash'=s.native_content_sha256 AND s.source_content_hash=s.native_content_sha256 AND e.source_snapshot_sha256=s.native_content_sha256 ORDER BY c.id,e.legacy_id FOR SHARE OF c,k,kb`, tenant, ns.KnowledgeBase, fmt.Sprint(id), nativeOrigin, ns.Knowledge, ns.Knowledge).Scan(&evidence).Error; err != nil { return err } attributes := []string{} unknown := false for _, row := range evidence { var source struct { Attributes json.RawMessage `json:"entity_attributes"` } if err := json.Unmarshal(row.Payload, &source); err != nil { return err } if len(source.Attributes) == 0 { unknown = true continue } var snapshot []*string if err := json.Unmarshal(source.Attributes, &snapshot); err != nil { return fmt.Errorf("invalid native graph entity attribute evidence: %w", err) } if snapshot == nil { unknown = true continue } values := make([]string, len(snapshot)) for i, value := range snapshot { if value == nil { return fmt.Errorf("invalid native graph entity attribute evidence: null attribute") } values[i] = *value } if len(values) == 0 { continue } if len(attributes) > 0 && !slices.Equal(attributes, values) { return fmt.Errorf("shared native graph entity attributes conflict; extraction requires review") } attributes = values } entityType := "concept" if len(attributes) > 0 { entityType = attributes[0] } if unknown { var existing Record result := tx.Raw(`SELECT payload FROM native_graph_records WHERE tenant_id=? AND knowledge_base_id=? AND source_table='knowledgeentity' AND legacy_id=? AND payload->>'origin'=? AND payload->>'native_identity'=?`, tenant, ns.KnowledgeBase, id, nativeOrigin, identity).Scan(&existing) if result.Error != nil { return result.Error } var prior struct { Attributes []string `json:"attributes"` Type string `json:"entity_type"` } if result.RowsAffected != 1 { return fmt.Errorf("shared native graph entity attribute provenance is unavailable; extraction requires review") } if err := json.Unmarshal(existing.Payload, &prior); err != nil { return err } // Legacy evidence cannot prove which surviving source owns a fact. // Refuse changing its aggregate until those sources are re-extracted. if !slices.Equal(prior.Attributes, attributes) || prior.Type != entityType { return fmt.Errorf("shared native graph entity attribute provenance is unavailable; extraction requires review") } attributes = prior.Attributes } payload["attributes"], payload["entity_type"] = attributes, entityType return nil } func putEvidence(tx *gorm.DB, tenant uint64, ns types.NameSpace, subject string, subjectID int64, chunk sourceChunk, quotation string, attributes ...[]string) error { id, identity := nativeID(ns.Knowledge, "evidence", subject, fmt.Sprint(subjectID), chunk.ID) excerpt := []rune(chunk.Content) if len(excerpt) > 2000 { excerpt = excerpt[:2000] } if quotation != "" { excerpt = []rune(quotation) } payload := map[string]any{"subject_type": subject, "subject_id": subjectID, "source_chunk_id": "native:" + chunk.ID, "source_file": chunk.FileName, "source_content_hash": contentHash(chunk.Content), "excerpt": string(excerpt)} if subject == "entity" { if len(attributes) != 1 { return fmt.Errorf("native graph entity evidence requires source attributes") } payload["entity_attributes"] = append([]string{}, attributes[0]...) } return putNativeRecord(tx, tenant, ns, "knowledgegraphevidence", id, identity, payload, contentHash(chunk.Content)) } func pruneNativeOrphans(tx *gorm.DB, tenant uint64, ns types.NameSpace) error { query := `DELETE FROM native_graph_records r WHERE r.tenant_id=? AND r.knowledge_base_id=? AND r.payload->>'origin'=? AND r.source_table IN ('knowledgeentity','knowledgerelation')` args := []any{tenant, ns.KnowledgeBase, nativeOrigin} if ns.Knowledge != "" { query += " AND r.payload->>'native_knowledge_id'=?" args = append(args, ns.Knowledge) } query += ` AND NOT EXISTS (SELECT 1 FROM native_graph_records e WHERE e.tenant_id=r.tenant_id AND e.knowledge_base_id=r.knowledge_base_id AND e.source_table='knowledgegraphevidence' AND e.payload->>'subject_id'=r.legacy_id::text AND e.payload->>'subject_type'=CASE r.source_table WHEN 'knowledgeentity' THEN 'entity' ELSE 'relation' END)` return tx.Exec(query, args...).Error } func (r *Repository) deleteNativeGraphs(ctx context.Context, namespaces []types.NameSpace) error { tenant, ok := types.TenantIDFromContext(ctx) if !ok || tenant == 0 || r.db == nil { return fmt.Errorf("graph deletion requires tenant and database") } for _, ns := range namespaces { if ns.KnowledgeBase == "" { return fmt.Errorf("graph deletion requires knowledge base scope") } } for _, ns := range namespaces { if err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if err := lockGraph(tx, tenant, ns.KnowledgeBase); err != nil { return err } query := tx.Table("native_graph_records").Where("tenant_id=? AND knowledge_base_id=? AND payload->>'origin'=?", tenant, ns.KnowledgeBase, nativeOrigin) if ns.Knowledge != "" { // Version copies preserve original payloads but remap source // pointers. Scope deletions through those current pointers too. query = query.Where("source_table IN ('knowledgegraphevidence','knowledgegraphchunkstate') AND (payload->>'native_knowledge_id'=? OR payload->>'source_chunk_id' IN (SELECT source_chunk_id FROM native_graph_sources WHERE tenant_id=? AND knowledge_base_id=? AND native_knowledge_id=?))", ns.Knowledge, tenant, ns.KnowledgeBase, ns.Knowledge) } if err := query.Delete(&Record{}).Error; err != nil { return err } if ns.Knowledge != "" { if err := pruneNativeOrphans(tx, tenant, types.NameSpace{KnowledgeBase: ns.KnowledgeBase}); err != nil { return err } } query = tx.Table("native_graph_sources").Where("tenant_id=? AND knowledge_base_id=? AND source_chunk_id LIKE 'native:%'", tenant, ns.KnowledgeBase) if ns.Knowledge != "" { query = query.Where("native_knowledge_id=?", ns.Knowledge) } return query.Delete(&Record{}).Error }); err != nil { return err } } if r.fallback != nil { return r.fallback.DelGraph(ctx, namespaces) } return nil }