Cyrene_v2/internal/runtime/memory/preinsert_store.go

package memory

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"
)

// 本文件是入库前审查（design/23 §15.1 D3）的存储层：候选读取、软合并落定、
// 回滚与审查留痕。它复用 memory 包既有的 *sql.DB、分区口径（tidyPartition）与
// 格式化辅助，不新建存储机制，也不碰任何既有行的 content（design §8）。
//
// 唯一的写入口径（§15 D3-6「merge/modify 一律走 §8 的既有口径」）：
//   - 被吸收的行：status='merged' + merged_into=<存活 id>；
//   - 存活行：merged_from 追加、merge_count +1、confidence 取较大者、
//     updated_at 刷新（§15 D3-2 T0-1 的「只更新 updated_at/confidence」）。
// 行**永不物理删除**，因此 revert 只需把 status 复原。

// ingestStatusMerged 是「已被合并」的行状态（design/23 §10 的 status 列）。
const ingestStatusMerged = "merged"

const listReviewCandidatesSQL = `SELECT id, persona_id, role_id, project_id, scope, kind,
	content, tags, confidence, source_conversation, created_at, updated_at,
	decay_weight FROM memory_items
	WHERE persona_id = ? AND scope = ? AND project_id = ? AND status = ? AND id <> ?
	ORDER BY created_at DESC, rowid DESC LIMIT ?`

const markAbsorbedSQL = `UPDATE memory_items
	SET status = ?, merged_into = ?, updated_at = ? WHERE id = ?`

const selectSurvivorMergeSQL = `SELECT merged_from, merge_count, confidence
	FROM memory_items WHERE id = ?`

const updateSurvivorSQL = `UPDATE memory_items
	SET merged_from = ?, merge_count = ?, confidence = ?, updated_at = ? WHERE id = ?`

const revertMergedSQL = `UPDATE memory_items
	SET status = ?, merged_into = '', updated_at = ? WHERE id = ? AND status = ?`

const selectMergedIntoSQL = `SELECT merged_into FROM memory_items
	WHERE id = ? AND status = ?`

const selectItemStatusSQL = `SELECT status FROM memory_items WHERE id = ?`

const insertIngestReviewSQL = `INSERT INTO memory_ingest_reviews
	(id, item_id, persona_id, scope, project_id, decision, reason, veto, target_id,
	 merged_ids, score, recall_mode, degraded, rule_version, session_id, created_at)
	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`

const listIngestReviewsSQL = `SELECT id, item_id, persona_id, scope, project_id, decision,
	reason, veto, target_id, merged_ids, score, recall_mode, degraded, rule_version,
	session_id, created_at FROM memory_ingest_reviews
	ORDER BY created_at DESC, rowid DESC LIMIT ?`

// AttachIngestReviewer 把入库前审查接到该 store 的写入路径：此后每条经 Add
// 落库的条目都会被投入审查队列（design §15 D3-1 的单一写入路径）。nil 是空操作
// （保持「未启用」的语义），可被并发调用，last-writer-wins。
func (s *Store) AttachIngestReviewer(r *PreInsertReviewer) {
	if s == nil || r == nil {
		return
	}
	s.reviewMu.Lock()
	defer s.reviewMu.Unlock()
	s.review = r
}

// ingestReviewer 返回当前接入的审查服务，未接入时为 nil。
func (s *Store) ingestReviewer() *PreInsertReviewer {
	if s == nil {
		return nil
	}
	s.reviewMu.RLock()
	defer s.reviewMu.RUnlock()
	return s.review
}

// enqueueReview 把刚落库的条目投入审查队列。非阻塞：队列满时审查方自己记
// WARN（它才是唯一知道队列状态的一侧），写入方绝不因此阻塞或失败。
func (s *Store) enqueueReview(item MemoryItem) {
	if r := s.ingestReviewer(); r != nil {
		r.Enqueue(item)
	}
}

// reviewItemStatus 读取一条记忆当前的在册状态。审查据此保证**每条记忆只审一次**：
// 已经被前一次审查合并掉的行再判一次会把存活行的 merge_count 重复加一
// （同一条事实被数两次），因此非 active 的行直接跳过。行不存在返回 ErrNotFound。
func (s *Store) reviewItemStatus(ctx context.Context, id string) (string, error) {
	var status string
	err := s.db.QueryRowContext(ctx, selectItemStatusSQL, id).Scan(&status)
	if errors.Is(err, sql.ErrNoRows) {
		return "", ErrNotFound
	}
	if err != nil {
		return "", fmt.Errorf("memory: read review item status: %w", err)
	}
	return status, nil
}

// listReviewCandidates 读取同一分区内的在册候选（最新在前）。分区过滤写在
// SQL 里，而不是读全表再在内存里筛 —— design §6 要求在分区切分层就隔断。
func (s *Store) listReviewCandidates(ctx context.Context, p tidyPartition, excludeID string, limit int) ([]MemoryItem, error) {
	if limit <= 0 {
		limit = defaultReviewScanLimit
	}
	rows, err := s.db.QueryContext(ctx, listReviewCandidatesSQL,
		p.personaID, string(p.scope), p.projectID, string(tidyStatusActive), excludeID, limit)
	if err != nil {
		return nil, fmt.Errorf("memory: list review candidates: %w", err)
	}
	return scanItems(rows, "review candidate")
}

// absorbIngestItem 把 absorbed 软合并进 survivor：一个事务里做「标记被吸收 +
// 刷新存活行的合并元信息」。两条 UPDATE 都只碰 status/merged_*/merge_count/
// confidence/updated_at，**绝不改 content**（design §8 的可回滚口径）。
func (s *Store) absorbIngestItem(ctx context.Context, absorbed, survivor MemoryItem, now time.Time) error {
	if absorbed.ID == "" || survivor.ID == "" || absorbed.ID == survivor.ID {
		return fmt.Errorf("memory: absorb ingest item: invalid pair %q -> %q", absorbed.ID, survivor.ID)
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("memory: absorb ingest item: begin: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	if _, err := tx.ExecContext(ctx, markAbsorbedSQL,
		string(ingestStatusMerged), survivor.ID, formatTime(now.UTC()), absorbed.ID); err != nil {
		return fmt.Errorf("memory: mark absorbed item: %w", err)
	}
	if err := bumpSurvivor(ctx, tx, survivor, absorbed, now); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("memory: absorb ingest item: commit: %w", err)
	}
	return nil
}

// bumpSurvivor 刷新存活行的合并元信息：merged_from 追加被吸收的 id、
// merge_count +1、confidence 取两者较大者、updated_at 刷新。
// 读-改-写在同一个事务内完成（连接池是单连接，天然串行），因此并发审查
// 不会互相覆盖。
func bumpSurvivor(ctx context.Context, tx *sql.Tx, survivor, absorbed MemoryItem, now time.Time) error {
	var (
		mergedFromJSON string
		mergeCount     int
		confidence     float64
	)
	if err := tx.QueryRowContext(ctx, selectSurvivorMergeSQL, survivor.ID).
		Scan(&mergedFromJSON, &mergeCount, &confidence); err != nil {
		return fmt.Errorf("memory: read survivor merge state: %w", err)
	}
	ids := make([]string, 0)
	if strings.TrimSpace(mergedFromJSON) != "" {
		if err := json.Unmarshal([]byte(mergedFromJSON), &ids); err != nil {
			return fmt.Errorf("memory: decode survivor merged_from: %w", err)
		}
	}
	if !containsIngestID(ids, absorbed.ID) {
		ids = append(ids, absorbed.ID)
	}
	encoded, err := json.Marshal(ids)
	if err != nil {
		return fmt.Errorf("memory: encode survivor merged_from: %w", err)
	}
	if absorbed.Confidence > confidence {
		// 置信度只升不降：同一条事实被再次观察到，是「它更可信」的证据，
		// 而不是要把它调低（§15 D3-2 T0-1 的「只更新 confidence」）。
		confidence = absorbed.Confidence
	}
	if _, err := tx.ExecContext(ctx, updateSurvivorSQL,
		string(encoded), mergeCount+1, confidence, formatTime(now.UTC()), survivor.ID); err != nil {
		return fmt.Errorf("memory: bump survivor item: %w", err)
	}
	return nil
}

// RevertMerged 撤销一次软合并（design/23 §8「回滚」）：把被吸收行的 status 复原
// 为 active 并清空 merged_into，同时把它的 id 从存活行的 merged_from 里摘掉、
// 把 merge_count 减回去（下限 1）。行与被合并的原文都还在，因此这是无损回滚。
// 该行不在 merged 状态时返回 ErrNotFound（不假装回滚成功）。
func (s *Store) RevertMerged(ctx context.Context, id string) error {
	if strings.TrimSpace(id) == "" {
		return ErrNotFound
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("memory: revert merged item: begin: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	var survivorID string
	err = tx.QueryRowContext(ctx, selectMergedIntoSQL, id, string(ingestStatusMerged)).Scan(&survivorID)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	if err != nil {
		return fmt.Errorf("memory: read merged item: %w", err)
	}
	now := time.Now().UTC()
	if _, err := tx.ExecContext(ctx, revertMergedSQL,
		string(tidyStatusActive), formatTime(now), id, string(ingestStatusMerged)); err != nil {
		return fmt.Errorf("memory: revert merged item: %w", err)
	}
	if survivorID != "" {
		if err := dropSurvivorLink(ctx, tx, survivorID, id, now); err != nil {
			return err
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("memory: revert merged item: commit: %w", err)
	}
	return nil
}

// dropSurvivorLink 把 id 从存活行的合并登记里摘掉：merged_from 去掉它、
// merge_count 减 1（下限 1）。存活行不存在（例如已被删除）时是空操作 ——
// 回滚的要点是「被吸收行复原」，绝不因为它而失败。
func dropSurvivorLink(ctx context.Context, tx *sql.Tx, survivorID, id string, now time.Time) error {
	var (
		mergedFromJSON string
		mergeCount     int
		confidence     float64
	)
	err := tx.QueryRowContext(ctx, selectSurvivorMergeSQL, survivorID).
		Scan(&mergedFromJSON, &mergeCount, &confidence)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("memory: read survivor for revert: %w", err)
	}
	ids := make([]string, 0)
	if strings.TrimSpace(mergedFromJSON) != "" {
		if err := json.Unmarshal([]byte(mergedFromJSON), &ids); err != nil {
			return fmt.Errorf("memory: decode survivor merged_from: %w", err)
		}
	}
	kept := make([]string, 0, len(ids))
	for _, v := range ids {
		if v != id {
			kept = append(kept, v)
		}
	}
	encoded, err := json.Marshal(kept)
	if err != nil {
		return fmt.Errorf("memory: encode survivor merged_from: %w", err)
	}
	next := mergeCount - 1
	if next < 1 {
		next = 1
	}
	if _, err := tx.ExecContext(ctx, updateSurvivorSQL,
		string(encoded), next, confidence, formatTime(now), survivorID); err != nil {
		return fmt.Errorf("memory: drop survivor link: %w", err)
	}
	return nil
}

// insertIngestReview 落一条审查留痕（design/23 §15 D3-8）。
func (s *Store) insertIngestReview(ctx context.Context, rev IngestReview) error {
	if rev.ID == "" {
		rev.ID = mustNewTidyID()
	}
	if rev.MergedIDs == nil {
		rev.MergedIDs = []string{}
	}
	mergedJSON, err := json.Marshal(rev.MergedIDs)
	if err != nil {
		return fmt.Errorf("memory: encode review merged ids: %w", err)
	}
	if rev.CreatedAt.IsZero() {
		rev.CreatedAt = time.Now().UTC()
	}
	if _, err := s.db.ExecContext(ctx, insertIngestReviewSQL,
		rev.ID, rev.ItemID, rev.PersonaID, string(rev.Scope), rev.ProjectID, string(rev.Decision),
		rev.Reason, string(rev.Veto), rev.TargetID, string(mergedJSON), rev.Score, rev.RecallMode,
		boolToInt(rev.Degraded), rev.RuleVer, rev.SessionID, formatTime(rev.CreatedAt)); err != nil {
		return fmt.Errorf("memory: insert ingest review: %w", err)
	}
	return nil
}

// ListIngestReviews 返回最近的审查留痕（最新在前，最多 limit 条；非正 limit
// 取 defaultReviewScanLimit）。读侧含命中 id 与依据，应由 owner-only 的
// 读侧使用（design §15 D3-8「面板可查」）。
func (s *Store) ListIngestReviews(ctx context.Context, limit int) ([]IngestReview, error) {
	if limit <= 0 {
		limit = defaultReviewScanLimit
	}
	rows, err := s.db.QueryContext(ctx, listIngestReviewsSQL, limit)
	if err != nil {
		return nil, fmt.Errorf("memory: list ingest reviews: %w", err)
	}
	defer func() { _ = rows.Close() }()

	out := make([]IngestReview, 0)
	for rows.Next() {
		rev, err := scanIngestReview(rows)
		if err != nil {
			return nil, fmt.Errorf("memory: scan ingest review: %w", err)
		}
		out = append(out, rev)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("memory: iterate ingest reviews: %w", err)
	}
	return out, nil
}

// scanIngestReview 读取一行审查留痕：merged_ids 以 JSON 数组存储，
// degraded 以 0/1 存储。
func scanIngestReview(sc rowScanner) (IngestReview, error) {
	var (
		rev             IngestReview
		scope, decision string
		veto            string
		mergedJSON      string
		degraded        int
		createdAt       string
	)
	if err := sc.Scan(&rev.ID, &rev.ItemID, &rev.PersonaID, &scope, &rev.ProjectID, &decision,
		&rev.Reason, &veto, &rev.TargetID, &mergedJSON, &rev.Score, &rev.RecallMode, &degraded,
		&rev.RuleVer, &rev.SessionID, &createdAt); err != nil {
		return IngestReview{}, err
	}
	rev.Scope = Scope(scope)
	rev.Decision = IngestDecision(decision)
	rev.Veto = TidyVetoReason(veto)
	rev.Degraded = degraded != 0
	if err := json.Unmarshal([]byte(mergedJSON), &rev.MergedIDs); err != nil {
		return IngestReview{}, fmt.Errorf("memory: decode review merged ids: %w", err)
	}
	if rev.MergedIDs == nil {
		rev.MergedIDs = []string{}
	}
	parsed, err := parseTime(createdAt)
	if err != nil {
		return IngestReview{}, err
	}
	rev.CreatedAt = parsed
	return rev, nil
}

// containsIngestID 报告 ids 里是否已有 id（避免 merged_from 重复登记同一条）。
func containsIngestID(ids []string, id string) bool {
	for _, v := range ids {
		if v == id {
			return true
		}
	}
	return false
}

// boolToInt 把布尔落成 SQLite 的 0/1。
func boolToInt(v bool) int {
	if v {
		return 1
	}
	return 0
}