Cyrene_v2/internal/runtime/history/refs.go

package history

// 本文件归属**文档引用计数**在消息写路径上的接线（design/24 §3.2 的「配额与清理」，
// 切片 S7）：引用来源 = chat_messages.document 列（≤5 个引用对象的 JSON 数组，
// 见 document.go），它在消息落库/编辑/删除时增减 documents.ref_count，并在被引用
// 时刷新 last_referenced_at。
//
// 三条纪律（它们就是「别四处散写」的落地方式）：
//
//  1. **增减点只有消息写路径**：insert（Append/AppendFull）、
//     UpdateMessageDocument（抽取结论回写）、DeleteMessage、DeleteSession；
//     别处一律不碰 documents 表。
//  2. **计数与消息行同生共死**：这些方法把引用计数的 SQL 放进**同一个事务**
//     （SQLExecer 由 *sql.Tx 满足），任何一边失败就整体回滚 —— 否则会出现
//     「消息在库里、计数没加」的窗口，而 S7 的 GC 恰恰是按计数判死的，
//     那等于让一份仍被引用的文件被回收。
//  3. **只按名字集合的差增减**：抽取结论回写常常是「同一批名字、只补
//     status/chars」（见 session_document_status.go），那一趟的增减必须为空；
//     否则同一份文档被重复计数，最终永远回收不掉。
//
// documents 表的具体 SQL 属 `internal/runtime/documents`（那张表的唯一写入面），
// 本包只定义**消费侧接口**并调用，因此依赖方向仍是 documents → history，不成环。

import (
	"context"
	"database/sql"
	"fmt"
)

// SQLExecer 是 *sql.DB 与 *sql.Tx 共同满足的最小 SQL 执行面。它让引用计数的
// 写入能落在调用方的**同一个事务**里（见本文件的三条纪律之 2），而不是另开连接
// 各自为政。
type SQLExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

// DocumentRefCounter 是文档引用计数的写入面，由 `internal/runtime/documents`
// 的 RefCounter 实现。names 是**次数敏感**的：同一个名字出现两次就代表两条引用，
// 实现方据此各增减一次。
//
// 它刻意不自己开事务：调用方（本包的消息写路径）把 ex 传进来，增减因此与消息行
// 原子落定。
type DocumentRefCounter interface {
	// Retain 为每个出现的名字记一次引用（+1，并刷新 last_referenced_at）。
	Retain(ctx context.Context, ex SQLExecer, names []string) error
	// Release 为每个出现的名字撤销一次引用（≤0 时钳到 0，绝不变成负数）。
	Release(ctx context.Context, ex SQLExecer, names []string) error
}

// StoreOption 配置 Store 的可选接线面（没有它时 Store 的行为与接线前逐字相同）。
type StoreOption func(*Store)

// WithDocumentRefs 注入文档引用计数写入面。未注入时引用计数不被维护（未接库的
// 构建/测试、以及只关心消息读写的调用方），本包对每次增减记一条 DEBUG 痕迹而
// 不是静默 —— 「没接线」与「接了但没记」必须能分辨。
func WithDocumentRefs(rc DocumentRefCounter) StoreOption {
	return func(s *Store) { s.refs = rc }
}

// documentNames 解出 document 列涉及的全部文档名。它**保留重复**（同一条消息
// 里同一份文档出现两次就是两次引用，insert 与 delete 两侧因此对称），解码失败时
// 返回包装了 ErrBadDocuments 的错误。
func documentNames(raw string) ([]string, error) {
	refs, err := DecodeDocuments(raw)
	if err != nil {
		return nil, err
	}
	names := make([]string, 0, len(refs))
	for _, ref := range refs {
		if ref.Name == "" {
			// EncodeDocuments 拒绝无名引用，因此这只可能来自外部改坏的行：
			// 名字是 URL 与展示身份的唯一来源，没有名字就没有可维护的引用。
			continue
		}
		names = append(names, ref.Name)
	}
	return names, nil
}

// retainDocuments 给 raw 里出现的每个文档名记一次引用。未注入计数器或列里没有
// 引用时是空操作。
func (s *Store) retainDocuments(ctx context.Context, ex SQLExecer, raw string) error {
	if s.refs == nil {
		s.log.Debug("document refs not wired: retain skipped", "documents", raw != "")
		return nil
	}
	names, err := documentNames(raw)
	if err != nil {
		// 名字解不出来就无从增减。这里**不**让消息写入失败：消息行才是用户的
		// 事实，而坏列的成因是外部改库/历史坏数据，不是一个可以在此修复的问题。
		// 代价是那份文档的计数可能偏低（保守方向：只会让它晚回收，绝不早删）。
		s.log.Warn("document refs retained skipped: malformed column", "err", err)
		return nil
	}
	if len(names) == 0 {
		return nil
	}
	if err := s.refs.Retain(ctx, ex, names); err != nil {
		return fmt.Errorf("history: retain document refs: %w", err)
	}
	return nil
}

// releaseDocuments 撤销 raw 里出现的每个文档名的一次引用。
func (s *Store) releaseDocuments(ctx context.Context, ex SQLExecer, raw string) error {
	if s.refs == nil {
		s.log.Debug("document refs not wired: release skipped", "documents", raw != "")
		return nil
	}
	names, err := documentNames(raw)
	if err != nil {
		s.log.Warn("document refs released skipped: malformed column", "err", err)
		return nil
	}
	if len(names) == 0 {
		return nil
	}
	if err := s.refs.Release(ctx, ex, names); err != nil {
		return fmt.Errorf("history: release document refs: %w", err)
	}
	return nil
}

// syncDocumentRefs 把一行消息的 document 列从 oldRaw 换成 newRaw 时按**差**增减
// 引用计数：新增的名字 retain、去掉的名字 release、两边都有的名字一动不动。两次
// 内容相同时（含都是空）它直接返回，因此「只补抽取结论」的回写不会重复计数。
func (s *Store) syncDocumentRefs(ctx context.Context, ex SQLExecer, oldRaw, newRaw string) error {
	if s.refs == nil || oldRaw == newRaw {
		return nil
	}
	oldNames, err := documentNames(oldRaw)
	if err != nil {
		s.log.Warn("document refs sync skipped: malformed old column", "err", err)
		return nil
	}
	newNames, err := documentNames(newRaw)
	if err != nil {
		s.log.Warn("document refs sync skipped: malformed new column", "err", err)
		return nil
	}
	add, remove := diffDocumentNames(oldNames, newNames)
	if len(add) > 0 {
		if err := s.refs.Retain(ctx, ex, add); err != nil {
			return fmt.Errorf("history: retain document refs: %w", err)
		}
	}
	if len(remove) > 0 {
		if err := s.refs.Release(ctx, ex, remove); err != nil {
			return fmt.Errorf("history: release document refs: %w", err)
		}
	}
	return nil
}

// diffDocumentNames 返回从 oldNames 到 newNames 的净增与净减（按出现次数计）。
// 输出的顺序是确定的（按各自的输入顺序），使同一次写入的 SQL 次序可复现、日志与
// 测试也稳定；两个列表里的同一个名字最多出现「次数差的绝对值」次。
func diffDocumentNames(oldNames, newNames []string) (add, remove []string) {
	remaining := make(map[string]int, len(oldNames))
	for _, name := range oldNames {
		remaining[name]++
	}
	for _, name := range newNames {
		if remaining[name] > 0 {
			remaining[name]--
			continue
		}
		add = append(add, name)
	}
	for _, name := range oldNames {
		if remaining[name] <= 0 {
			continue
		}
		remaining[name]--
		remove = append(remove, name)
	}
	return add, remove
}