knowledge-governance/internal/releaseadapter/v071/library_production_target.go

package v071

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"knowledge-governance/internal/libraryrelease"
	"knowledge-governance/internal/librarysnapshot"
)

// 生产整库目标：真实读、真实写、真实校验、真实恢复。
//
// 只覆盖知识域（文档/分块/标签/资源绑定/Wiki/资源对象/检索索引），不写 users、
// sessions、api_keys 这类账号与会话表；索引直接搬运归档里的原始文档与向量，
// 不重新解析、不重新 Embedding。连接全部来自注入配置，代码里不存生产凭据。

var ErrProductionTargetUnconfigured = errors.New("生产整库目标未配置（需要生产数据库、索引、对象存储与知识库映射）")

// ProductionScope 是一个目标库在生产侧的身份。
type ProductionScope struct {
	TenantID        uint64
	KnowledgeBaseID string
}

type ProductionLibraryConfig struct {
	// Postgres 是生产 WeKnora 的数据库连接串（部署注入，不落代码/日志）。
	Postgres string
	// SearchURL 是生产 Elasticsearch 基址；SearchIndex 是知识域索引名。
	SearchURL   string
	SearchIndex string
	SearchToken string
	// Objects 负责写入资源对象（既有 ObjectStore 接口）；ObjectsReader 负责
	// 备份时读回对象字节。同一个 S3 存储通常两者都实现。
	Objects       ObjectStore
	ObjectsReader ObjectReader
	ArchiveRoot   string
	// ObjectsBucket 是目标对象存储的桶名：资源行里的定位符要换成它，
	// 否则同步过去的图/附件仍指向来源桶。
	ObjectsBucket string
	// ObjectsBackendID 是目标实例 storage_backends 的主键。留空则按租户+桶名在
	// 目标库里解析；定位符里的来源后端 id 在目标实例不存在，原版应用会 500。
	ObjectsBackendID string
	MaxDocuments     int
	HTTPClient       *http.Client
	Now              func() time.Time
}

func (config ProductionLibraryConfig) validate(scope ProductionScope) error {
	if config.Postgres == "" || config.SearchURL == "" || config.SearchIndex == "" ||
		config.Objects == nil || config.ObjectsReader == nil || config.ArchiveRoot == "" || config.MaxDocuments < 1 {
		return ErrProductionTargetUnconfigured
	}
	if scope.TenantID == 0 || !validSearchIdentifier(scope.KnowledgeBaseID) {
		return fmt.Errorf("%w：缺少目标知识库身份", ErrProductionTargetUnconfigured)
	}
	return nil
}

// 依赖顺序：插入按正序，删除按倒序。
// 真实 Schema 有外键（resource_bindings.resource_id → resources.id），
// 所以父表 resources 必须排在子表 resource_bindings 之前。
var productionLibraryTables = []string{
	"knowledges", "chunks", "knowledge_tags", "knowledge_tag_relations",
	"resources", "resource_bindings", "wiki_folders", "wiki_pages",
	"wiki_page_issues", "wiki_log_entries",
}

type ProductionLibraryTarget struct {
	pool   *pgxpool.Pool
	config ProductionLibraryConfig
	scope  ProductionScope
	search *SearchExporter
	now    func() time.Time
	// storageBackend / storageProvider 是本次写入使用的目标存储后端与提供方
	// （写入与校验共用，缺一不可：只换桶不换提供方，目标取文件仍会走来源实例的路）。
	storageBackend  string
	storageProvider string
	// snapshotTenant / snapshotKB 是这次要写入的快照来源身份。写入时按它做
	// 来源→目标的范围映射；校验时按它反查"目标里还有没有来源范围的行"。
	snapshotTenant uint64
	snapshotKB     string
	// targetEmbeddingModel 是目标知识库绑定的 embedding 模型 id（文档行要换成它）。
	targetEmbeddingModel string
}

// resolveStorageBackend 在目标库里找本次写入的存储后端。找不到就报错：宁可不写，
// 也不写一条目标应用读不出来的定位符（上轮就是这样同步"成功"了却下载 500）。
func (t *ProductionLibraryTarget) resolveStorageBackend(ctx context.Context) error {
	if t.config.ObjectsBucket == "" {
		return nil
	}
	if t.config.ObjectsBackendID != "" {
		if !validSearchIdentifier(t.config.ObjectsBackendID) {
			return errors.New("目标存储后端 id 形状不合法")
		}
		provider, err := t.storageProviderOf(ctx, t.config.ObjectsBackendID)
		if err != nil {
			return err
		}
		t.storageBackend, t.storageProvider = t.config.ObjectsBackendID, provider
		return nil
	}
	var id, provider string
	err := t.pool.QueryRow(ctx, `
SELECT id, coalesce(provider,'') FROM storage_backends
WHERE tenant_id=$1 AND coalesce(config->>'bucket_name','')=$2
ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id, &provider)
	if err != nil {
		return errors.New("目标实例没有与目标桶匹配的存储后端，拒绝写入不可解析的资源定位符")
	}
	if !validSearchIdentifier(id) {
		return errors.New("目标存储后端 id 形状不合法")
	}
	t.storageBackend, t.storageProvider = id, provider
	return nil
}

// storageProviderOf 读出目标存储后端的提供方（minio / oss / s3…）。资源行里的
// 提供方必须与它一致，否则目标应用会按来源提供方去找一个不存在的存储服务。
func (t *ProductionLibraryTarget) storageProviderOf(ctx context.Context, id string) (string, error) {
	var provider string
	if err := t.pool.QueryRow(ctx, `SELECT coalesce(provider,'') FROM storage_backends WHERE id=$1 AND tenant_id=$2`,
		id, t.scope.TenantID).Scan(&provider); err != nil {
		return "", errors.New("目标实例没有这个存储后端，拒绝写入不可解析的资源定位符")
	}
	return provider, nil
}

func NewProductionLibraryTarget(ctx context.Context, config ProductionLibraryConfig, scope ProductionScope) (*ProductionLibraryTarget, error) {
	if err := config.validate(scope); err != nil {
		return nil, err
	}
	pool, err := pgxpool.New(ctx, config.Postgres)
	if err != nil {
		// 连接串本身不打印，错误里也只说明连不上。
		return nil, errors.New("生产数据库连接失败")
	}
	client := config.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: 2 * time.Minute}
	}
	search, err := NewSearchExporter(SearchExporterConfig{BaseURL: config.SearchURL, Index: config.SearchIndex,
		AuthorizationToken: config.SearchToken, PageSize: 500, HTTPClient: client})
	if err != nil {
		pool.Close()
		return nil, err
	}
	now := config.Now
	if now == nil {
		now = time.Now
	}
	return &ProductionLibraryTarget{pool: pool, config: config, scope: scope, search: search, now: now}, nil
}

func (t *ProductionLibraryTarget) Close() { t.pool.Close() }

// Pool/Search/Identity 暴露给预备份存储复用同一套生产连接，避免重复开连接池。
func (t *ProductionLibraryTarget) Pool() *pgxpool.Pool { return t.pool }

func (t *ProductionLibraryTarget) Search() *SearchExporter { return t.search }

func (t *ProductionLibraryTarget) Objects() ObjectReader { return t.config.ObjectsReader }

func (t *ProductionLibraryTarget) Identity() PostgresIdentity {
	return PostgresIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}
}

// Current 读生产当前整库指纹：每张表的 (行数, 身份集合摘要) + 索引条数。
func (t *ProductionLibraryTarget) Current(ctx context.Context) (libraryrelease.CurrentState, error) {
	state := libraryrelease.CurrentState{Counts: map[string]int64{}, Digests: map[string]string{},
		CapturedAt: t.now().UTC()}
	for _, table := range productionLibraryTables {
		count, digest, err := t.tableState(ctx, table)
		if err != nil {
			return libraryrelease.CurrentState{}, err
		}
		state.Counts[table] = count
		state.Digests[table] = digest
	}
	indexCount, err := t.search.Count(ctx, SearchIdentity{TenantID: t.scope.TenantID,
		KnowledgeBaseID: t.scope.KnowledgeBaseID})
	if errors.Is(err, errIndexMissing) {
		// 生产索引不存在 = 空基线（零文档）；不是故障，但也不会被当成"有内容"。
		indexCount = 0
	} else if err != nil {
		return libraryrelease.CurrentState{}, err
	}
	state.Counts["index"] = indexCount
	return state, nil
}

// scopeWhere 返回"这张表里属于本知识库的行"的过滤条件（$1=tenant，$2=kb）。
// 不同表的归属列不同：关系表与绑定表没有 knowledge_base_id，必须走父表判定，
// 否则要么删错、要么把别的知识库数据算进来。
func scopeWhere(table string) string {
	switch table {
	case "knowledge_tag_relations":
		// 真实 WeKnora 的这张表只有 knowledge_id/tag_id/created_at，
		// 没有 tenant_id/knowledge_base_id：归属只能走父表 knowledges 判定。
		return `knowledge_id IN (SELECT id FROM knowledges k
 WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)`
	case "resource_bindings":
		return `tenant_id=$1 AND ((owner_type='knowledge' AND owner_id IN (SELECT id FROM knowledges k
 WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)) OR (owner_type='chunk' AND owner_id IN
 (SELECT id FROM chunks c WHERE c.tenant_id=$1 AND c.knowledge_base_id=$2)))`
	case "resources":
		// 必须把外层 resources.id 与绑定关系精确关联：只按 owner_id 判断会命中整张
		// 资源表；同时要覆盖 chunk 绑定（图片/附件常只挂在分块上）并限制绑定租户。
		// 两种归属都要算：显式绑定（knowledge/chunk）+ 被本库 chunk 正文以
		// resource://<handle> 引用（图片/附件常见，可能没有绑定行）。
		// 与归档侧收集资源部件的规则保持一致，否则校验会把导入的资源算成"多余"。
		return `(EXISTS (
 SELECT 1 FROM resource_bindings b
 JOIN knowledges k ON k.id=b.owner_id AND k.tenant_id=b.tenant_id
 WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='knowledge' AND k.knowledge_base_id=$2
 UNION ALL
 SELECT 1 FROM resource_bindings b
 JOIN chunks c ON c.id=b.owner_id AND c.tenant_id=b.tenant_id
 WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='chunk' AND c.knowledge_base_id=$2)
 OR EXISTS (
 SELECT 1 FROM chunks c
 WHERE c.tenant_id=$1 AND c.knowledge_base_id=$2 AND c.deleted_at IS NULL
   AND coalesce(resources.handle,'') <> ''
   AND position('resource://'||resources.handle IN coalesce(c.content,'')) > 0))`
	default:
		return "tenant_id=$1 AND knowledge_base_id=$2"
	}
}

// errMissingIdentity 表示归档里的这张表没有代理主键，只能靠内容摘要核对。
var errMissingIdentity = errors.New("missing stable identity")

// identityExpr 是"稳定身份"列：多数表就是 id，关系表没有 id 就用复合键。
func identityExpr(table string) string {
	if table == "knowledge_tag_relations" {
		return "knowledge_id||':'||tag_id"
	}
	return "id"
}

func (t *ProductionLibraryTarget) tableState(ctx context.Context, table string) (int64, string, error) {
	rows, err := t.pool.Query(ctx, fmt.Sprintf(`SELECT %s FROM %s WHERE %s ORDER BY 1`,
		identityExpr(table), table, scopeWhere(table)), t.scope.TenantID, t.scope.KnowledgeBaseID)
	if err != nil {
		return 0, "", fmt.Errorf("读取生产 %s 失败", table)
	}
	defer rows.Close()
	ids := make([]string, 0, 64)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return 0, "", err
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return 0, "", err
	}
	return int64(len(ids)), digestIDs(ids), nil
}

func digestIDs(ids []string) string {
	sum := sha256.Sum256([]byte(strings.Join(ids, "\n")))
	return hex.EncodeToString(sum[:])
}

// Apply 把一类数据从归档写进生产；database 与 wiki 走同一个知识域事务。
func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryrelease.Stage, snapshotID string) error {
	if err := t.resolveStorageBackend(ctx); err != nil {
		return err
	}
	t.resolveEmbeddingModel(ctx)
	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
	if err != nil {
		return errors.New("归档不可用")
	}
	defer archive.Close()
	manifest, err := archive.Load(snapshotID)
	if err != nil {
		return errors.New("预发布快照不可读")
	}
	if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
		return err
	}
	switch stage {
	case libraryrelease.StageDatabase, libraryrelease.StageWiki:
		kind := "database"
		if stage == libraryrelease.StageWiki {
			kind = "wiki"
		}
		return t.importRows(ctx, archive, manifest, kind)
	case libraryrelease.StageResources:
		return t.importResources(ctx, archive, manifest)
	case libraryrelease.StageIndex:
		return t.importIndex(ctx, archive, manifest)
	default:
		return fmt.Errorf("未知阶段 %q", stage)
	}
}

func (t *ProductionLibraryTarget) importRows(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest, kind string) error {
	parts := map[string]librarysnapshot.Part{}
	for _, part := range manifest.Parts {
		if part.Kind == kind {
			parts[part.Name] = part
		}
	}
	if len(parts) == 0 {
		return fmt.Errorf("快照缺少 %s 部件", kind)
	}
	tx, err := t.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return errors.New("开启生产事务失败")
	}
	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
	for index := len(productionLibraryTables) - 1; index >= 0; index-- {
		table := productionLibraryTables[index]
		if _, ok := parts[table]; !ok {
			continue
		}
		if err := t.deleteScope(ctx, tx, table); err != nil {
			return err
		}
	}
	for _, table := range productionLibraryTables {
		part, ok := parts[table]
		if !ok {
			continue
		}
		if err := t.insertRows(ctx, tx, archive, part, table); err != nil {
			return err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return errors.New("提交生产写入失败")
	}
	return nil
}

func (t *ProductionLibraryTarget) deleteScope(ctx context.Context, tx pgx.Tx, table string) error {
	if table == "resources" {
		// 资源可能被多个知识库共享：这里不删，只做插入/更新。
		return nil
	}
	statement := fmt.Sprintf(`DELETE FROM %s WHERE %s`, table, scopeWhere(table))
	if _, err := tx.Exec(ctx, statement, t.scope.TenantID, t.scope.KnowledgeBaseID); err != nil {
		return fmt.Errorf("清理生产 %s 失败", table)
	}
	return nil
}

func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, archive *librarysnapshot.Archive, part librarysnapshot.Part, table string) error {
	reader, err := archive.OpenPart(part)
	if err != nil {
		return fmt.Errorf("读取归档 %s 失败", part.Name)
	}
	defer reader.Close()
	decoder := json.NewDecoder(reader)
	token, err := decoder.Token()
	if err != nil || token != json.Delim('[') {
		return fmt.Errorf("归档 %s 不是行数组", part.Name)
	}
	var columns []string
	statement := ""
	for decoder.More() {
		var row map[string]json.RawMessage
		if err := decoder.Decode(&row); err != nil {
			return fmt.Errorf("解析归档 %s 行失败", part.Name)
		}
		if len(row) == 0 {
			continue
		}
		// 行必须先映射到目标范围再落库：原样插入会把行写进来源租户/知识库，
		// 目标范围的删除、校验与恢复都看不见它们。
		if err := t.retargetRow(table, row); err != nil {
			return err
		}
		if columns == nil {
			columns = make([]string, 0, len(row))
			for key := range row {
				columns = append(columns, key)
			}
			sort.Strings(columns)
			// 本库行先删后插：除 resources（可能被多个知识库共享）外不需要 upsert。
			if table == "resources" {
				statement = fmt.Sprintf(`INSERT INTO %s SELECT * FROM jsonb_populate_record(NULL::%s, $1::jsonb)
ON CONFLICT (id) DO UPDATE SET %s`, table, table, updateAssignments(columns))
			} else {
				statement = fmt.Sprintf(`INSERT INTO %s SELECT * FROM jsonb_populate_record(NULL::%s, $1::jsonb)`,
					table, table)
			}
		}
		encoded, err := json.Marshal(row)
		if err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, statement, encoded); err != nil {
			return fmt.Errorf("写入生产 %s 失败: %w", table, err)
		}
	}
	return nil
}

func updateAssignments(columns []string) string {
	assignments := make([]string, 0, len(columns))
	for _, column := range columns {
		if column == "id" {
			continue
		}
		assignments = append(assignments, fmt.Sprintf("%s=EXCLUDED.%s", column, column))
	}
	if len(assignments) == 0 {
		return "id=EXCLUDED.id"
	}
	return strings.Join(assignments, ", ")
}

// resourceLocation 是资源对象在生产存储里的位置与提供方（oss / s3 / minio…）。
type resourceLocation struct {
	Path     string
	Provider string
}

// rowTransform 是"写入时对行做过的变换"的统一入口：校验要用同一份，否则摘要比的是
// 变换前的行、目标里是变换后的行，永远不一致。范围映射同样必须两边一致。
func (t *ProductionLibraryTarget) rowTransform(table string) func(map[string]json.RawMessage) error {
	return func(row map[string]json.RawMessage) error { return t.retargetRow(table, row) }
}

func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
	paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
	if err != nil {
		return err
	}
	for _, part := range manifest.Parts {
		if part.Kind != "resource" {
			continue
		}
		location, ok := paths[part.Name]
		if !ok || location.Path == "" {
			return fmt.Errorf("生产缺少资源 %s 的存储路径", part.Name)
		}
		reader, err := archive.OpenPart(part)
		if err != nil {
			return err
		}
		body, err := io.ReadAll(io.LimitReader(reader, part.Bytes+1))
		reader.Close()
		if err != nil || int64(len(body)) != part.Bytes {
			return fmt.Errorf("资源 %s 字节与归档摘要不符", part.Name)
		}
		if err := t.config.Objects.Put(ctx, location.Provider, location.Path, bytes.NewReader(body), int64(len(body))); err != nil {
			return fmt.Errorf("上传资源 %s 失败: %w", part.Name, err)
		}
	}
	return nil
}

// verifyObjects 逐个比对生产对象的实际字节摘要与归档摘要：
// 只有"对象真的在、内容一致"才算通过，防止对象丢失或损坏被行数掩盖。
func (t *ProductionLibraryTarget) verifyObjects(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
	paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
	if err != nil {
		return err
	}
	for _, part := range manifest.Parts {
		if part.Kind != "resource" {
			continue
		}
		location, ok := paths[part.Name]
		if !ok || location.Path == "" {
			return fmt.Errorf("生产缺少资源 %s", part.Name)
		}
		reader, err := t.config.ObjectsReader.Open(ctx, location.Provider, location.Path)
		if err != nil {
			return fmt.Errorf("读取生产资源 %s 失败: %w", part.Name, err)
		}
		hash := sha256.New()
		size, err := io.Copy(hash, io.LimitReader(reader, part.Bytes+1))
		reader.Close()
		if err != nil {
			return fmt.Errorf("读取生产资源 %s 失败", part.Name)
		}
		if size != part.Bytes || hex.EncodeToString(hash.Sum(nil)) != part.SHA256 {
			return fmt.Errorf("生产资源 %s 与归档摘要不一致（%d 字节）", part.Name, size)
		}
	}
	return nil
}

func resourcePartNames(manifest librarysnapshot.Manifest) []string {
	names := make([]string, 0, 16)
	for _, part := range manifest.Parts {
		if part.Kind == "resource" {
			names = append(names, part.Name)
		}
	}
	return names
}

// resourcePaths 只在"本次要写的这些资源 ID"里解析路径，并且强制租户。
// 不按绑定范围扫全表：被 chunk 正文以 resource:// 引用的资源可能没有绑定行，
// 但它是这份快照的一部分；反过来按绑定扫会漏掉它们，按全表扫会读到别的库。
func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context, ids []string) (map[string]resourceLocation, error) {
	if len(ids) == 0 {
		return map[string]resourceLocation{}, nil
	}
	rows, err := t.pool.Query(ctx, `SELECT id, coalesce(physical_path,''), coalesce(provider,'') FROM resources
WHERE tenant_id=$1 AND id = ANY($2::text[])`, t.scope.TenantID, ids)
	if err != nil {
		return nil, fmt.Errorf("读取生产资源路径失败: %w", err)
	}
	defer rows.Close()
	paths := map[string]resourceLocation{}
	for rows.Next() {
		var id, path, provider string
		if err := rows.Scan(&id, &path, &provider); err != nil {
			return nil, err
		}
		paths[id] = resourceLocation{Path: path, Provider: provider}
	}
	return paths, rows.Err()
}

func (t *ProductionLibraryTarget) importIndex(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
	for _, part := range manifest.Parts {
		if part.Kind != "index" {
			continue
		}
		if part.Count > int64(t.config.MaxDocuments) {
			return fmt.Errorf("索引条数 %d 超过上限 %d", part.Count, t.config.MaxDocuments)
		}
		reader, err := archive.OpenPart(part)
		if err != nil {
			return err
		}
		documents, err := librarysnapshot.ReadIndexDocuments(reader)
		reader.Close()
		if err != nil {
			return err
		}
		// 直接搬运原始文档与向量：不解析、不重新 Embedding。
		identity := SearchIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}
		if err := t.search.BulkIndex(ctx, documents, identity); err != nil {
			return err
		}
		// 快照里没有的文档必须从生产索引删掉，否则撤销只在数据库生效。
		keep := make([]string, 0, len(documents))
		for _, document := range documents {
			keep = append(keep, document.ID)
		}
		if _, err := t.search.PruneScope(ctx, identity, keep); err != nil {
			return err
		}
	}
	return nil
}

// Verify 用行数 + 身份集合摘要 + 索引条数核对生产与预发布快照是否一致。
func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrelease.Expectation) error {
	if err := t.resolveStorageBackend(ctx); err != nil {
		return err
	}
	t.resolveEmbeddingModel(ctx)
	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
	if err != nil {
		return errors.New("归档不可用")
	}
	defer archive.Close()
	manifest, err := archive.Load(expect.SnapshotID)
	if err != nil {
		return errors.New("预发布快照不可读")
	}
	if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
		return err
	}
	for _, part := range manifest.Parts {
		if part.Kind != "database" && part.Kind != "wiki" {
			continue
		}
		var resourceIDs []string
		if part.Name == "resources" {
			resourceIDs, err = archivedResourceIDs(archive, part)
			if err != nil {
				return err
			}
		}
		wantCount, wantDigest, err := archivedTableState(archive, part)
		if errors.Is(err, errMissingIdentity) {
			// 个别表（例如日志表）没有代理主键：身份集合无从比，但行数与逐行内容摘要仍然必须一致。
			wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part, t.rowTransform(part.Name))
			if contentErr != nil {
				return contentErr
			}
			contentCount, contentDigest, contentErr := t.productionTableContent(ctx, part.Name, columns)
			if contentErr != nil {
				return contentErr
			}
			if contentCount != wantContentCount || contentDigest != wantContentDigest {
				return fmt.Errorf("生产 %s 内容与快照不一致：%d 行", part.Name, contentCount)
			}
			continue
		}
		if err != nil {
			return err
		}
		count, digest, err := t.tableState(ctx, part.Name)
		if part.Name == "resources" {
			// resources 是多知识库共享表，按"本库可见"重新推导会和采集口径打架
			// （采集看 knowledges/chunks/wiki 正文里的 handle，校验只看 chunks），
			// 于是恢复得一字不差也会被判失败。改成按归档的身份集合核对：
			// 归档里每一行都必须在、内容摘要也一致，缺一行就失败。
			count, digest, err = t.archivedResourceState(ctx, resourceIDs)
		}
		if err != nil {
			return err
		}
		if count != wantCount || digest != wantDigest {
			return fmt.Errorf("生产 %s 与快照不一致：行数 %d/%d", part.Name, count, wantCount)
		}
		// 行数/身份一致还要比内容：同 ID 改正文、改 Wiki、改向量必须被发现。
		wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part, t.rowTransform(part.Name))
		if err != nil {
			return err
		}
		contentCount, contentDigest := int64(0), ""
		if part.Name == "resources" {
			contentCount, contentDigest, err = t.productionResourceContent(ctx, resourceIDs, columns)
		} else {
			contentCount, contentDigest, err = t.productionTableContent(ctx, part.Name, columns)
		}
		if err != nil {
			return err
		}
		if contentCount != wantContentCount || contentDigest != wantContentDigest {
			return fmt.Errorf("生产 %s 内容与快照不一致：%d 行 / 摘要不一致", part.Name, contentCount)
		}
	}
	if err := t.verifyIndexDigest(ctx, archive, manifest); err != nil {
		return err
	}
	if err := t.verifyObjects(ctx, archive, manifest); err != nil {
		return err
	}
	// 目标里不允许留下"快照来源范围"的行：那说明有写入逃出了目标范围，
	// 而恢复只按目标范围回放，永远删不到它们（真实事故就是这样被漏掉的）。
	if err := t.assertNoSourceScopeRows(ctx, archive, manifest); err != nil {
		return err
	}
	return nil
}

// dropOrphanResources 清掉"本次恢复后已无任何归属"的目标租户资源行。
//
// 判据必须同时满足：属于目标租户、不在预备份快照里、没有任何 resource_binding、
// 也没有任何分块正文以 resource://<handle> 引用它。这样的行对目标应用完全不可见
// （没有知识库能按它取到文件），只可能是失败写入留下的残渣。
func (t *ProductionLibraryTarget) dropOrphanResources(ctx context.Context, backupSnapshotID string) error {
	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
	if err != nil {
		return errors.New("归档不可用")
	}
	defer archive.Close()
	manifest, err := archive.Load(backupSnapshotID)
	if err != nil {
		return errors.New("预备份快照不可读")
	}
	kept := []string{}
	for _, part := range manifest.Parts {
		if part.Kind != "database" || part.Name != "resources" {
			continue
		}
		ids, err := archivedResourceIDs(archive, part)
		if err != nil {
			return err
		}
		kept = append(kept, ids...)
	}
	rows, err := t.pool.Query(ctx, `
SELECT r.id FROM resources AS r
WHERE r.tenant_id=$1
  AND NOT (r.id = ANY($2::text[]))
  AND NOT EXISTS (SELECT 1 FROM resource_bindings b WHERE b.resource_id=r.id)
  AND NOT EXISTS (SELECT 1 FROM chunks c WHERE c.tenant_id=r.tenant_id AND c.deleted_at IS NULL
                  AND coalesce(r.handle,'') <> ''
                  AND position('resource://'||r.handle IN coalesce(c.content,'')) > 0)`,
		t.scope.TenantID, kept)
	if err != nil {
		return fmt.Errorf("查找目标残留资源失败: %w", err)
	}
	orphans := make([]string, 0, 8)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return err
		}
		orphans = append(orphans, id)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}
	if _, err := t.pool.Exec(ctx, `DELETE FROM resources WHERE tenant_id=$1 AND id = ANY($2::text[])`,
		t.scope.TenantID, orphans); err != nil {
		return fmt.Errorf("清理目标残留资源失败: %w", err)
	}
	return nil
}

// archivedResourceState 只核对"归档里列出的这些资源"是否还在且未变。
func (t *ProductionLibraryTarget) archivedResourceState(ctx context.Context, ids []string) (int64, string, error) {
	rows, err := t.pool.Query(ctx, `SELECT id FROM resources WHERE id = ANY($1::text[]) ORDER BY id`, ids)
	if err != nil {
		return 0, "", errors.New("读取生产 resources 失败")
	}
	defer rows.Close()
	found := make([]string, 0, len(ids))
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return 0, "", err
		}
		found = append(found, id)
	}
	if err := rows.Err(); err != nil {
		return 0, "", err
	}
	return int64(len(found)), digestIDs(found), nil
}

func archivedResourceIDs(archive *librarysnapshot.Archive, part librarysnapshot.Part) ([]string, error) {
	reader, err := archive.OpenPart(part)
	if err != nil {
		return nil, err
	}
	defer reader.Close()
	decoder := json.NewDecoder(reader)
	token, err := decoder.Token()
	if err != nil || token != json.Delim('[') {
		return nil, errors.New("归档部件不是行数组")
	}
	ids := make([]string, 0, 64)
	for decoder.More() {
		var row map[string]json.RawMessage
		if err := decoder.Decode(&row); err != nil {
			return nil, err
		}
		var id string
		if err := json.Unmarshal(row["id"], &id); err != nil || id == "" {
			return nil, errors.New("归档资源缺少身份")
		}
		ids = append(ids, id)
	}
	sort.Strings(ids)
	return ids, nil
}

func archivedTableState(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, error) {
	reader, err := archive.OpenPart(part)
	if err != nil {
		return 0, "", err
	}
	defer reader.Close()
	decoder := json.NewDecoder(reader)
	token, err := decoder.Token()
	if err != nil || token != json.Delim('[') {
		return 0, "", errors.New("归档部件不是行数组")
	}
	ids := make([]string, 0, 64)
	for decoder.More() {
		var row map[string]json.RawMessage
		if err := decoder.Decode(&row); err != nil {
			return 0, "", err
		}
		identity := archiveRowIdentity(part.Name, row)
		if identity == "" {
			return 0, "", errMissingIdentity
		}
		ids = append(ids, identity)
	}
	sort.Strings(ids)
	return int64(len(ids)), digestIDs(ids), nil
}

func archiveRowIdentity(table string, row map[string]json.RawMessage) string {
	value := func(key string) string {
		var text string
		_ = json.Unmarshal(row[key], &text)
		return text
	}
	if table == "knowledge_tag_relations" {
		knowledge, tag := value("knowledge_id"), value("tag_id")
		if knowledge == "" || tag == "" {
			return ""
		}
		return knowledge + ":" + tag
	}
	return value("id")
}

// Restore 用预备份把生产整库恢复回去：走同一条 Apply 路径，可重复执行。
func (t *ProductionLibraryTarget) Restore(ctx context.Context, backup libraryrelease.BackupRef) error {
	if backup.SnapshotID == "" {
		return errors.New("恢复需要绑定的预备份快照")
	}
	for _, stage := range libraryrelease.Stages() {
		if err := t.Apply(ctx, stage, backup.SnapshotID); err != nil {
			return fmt.Errorf("恢复阶段 %s 失败: %w", stage, err)
		}
	}
	// resources 是多知识库共享表，回放阶段不删它：本次写入新增、恢复后已经
	// 没有任何归属的资源行会残留。它们对应用不可见，但属于"多写的"。
	if err := t.dropOrphanResources(ctx, backup.SnapshotID); err != nil {
		return err
	}
	// 恢复工具把四个阶段都跑完不算恢复完成：整实例必须回到预备份时刻的样子。
	// 只比目标范围会把"写到范围外"的残留当成已恢复（真实事故：恢复后仍多 67 行）。
	if len(backup.Universe) > 0 {
		after, err := universeFingerprint(ctx, t.pool)
		if err != nil {
			return fmt.Errorf("恢复后读取整库指纹失败: %w", err)
		}
		if diff := universeDiff(backup.Universe, after); diff != "" {
			return fmt.Errorf("恢复后目标与预备份时刻不一致：%s（很可能有写入逃出了预备份范围，不能当作已恢复）", diff)
		}
	}
	return nil
}