feat(search): add transactional rebuild and index optimization

internal/search/maintenance.go

@@ -0,0 +1,99 @@
+package search
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"rawmark/internal/vault"
+)
+
+func insert(tx *sql.Tx, p, rev string, content []byte) (int64, error) {
+	metadata, err := json.Marshal(vault.Analyze(p, content))
+	if err != nil {
+		return 0, err
+	}
+	result, err := tx.Exec(`INSERT INTO docs(path,rev,metadata) VALUES (?,?,?)`, p, rev, string(metadata))
+	if err != nil {
+		return 0, err
+	}
+	id, err := result.LastInsertId()
+	if err != nil {
+		return 0, err
+	}
+	_, err = tx.Exec(`INSERT INTO docs_fts(rowid,path,body) VALUES (?,?,?)`, id, p, string(content))
+	return id, err
+}
+
+// Rebuild 与后台同步互斥，失败时事务和已知快照均保持原状。
+func (x *Index) Rebuild(ctx context.Context) error {
+	x.work.Lock()
+	defer x.work.Unlock()
+	x.mu.Lock()
+	version := x.pendingVersion
+	x.mu.Unlock()
+	entries, err := x.v.Tree()
+	if err != nil {
+		return err
+	}
+	tx, err := x.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if _, err = tx.ExecContext(ctx, `DELETE FROM docs_fts; DELETE FROM docs`); err != nil {
+		return err
+	}
+	next := map[string]doc{}
+	for _, entry := range entries {
+		if entry.Dir || !vault.IsNote(entry.Path) {
+			continue
+		}
+		if err = ctx.Err(); err != nil {
+			return err
+		}
+		content, rev, err := x.v.Read(entry.Path)
+		if err != nil {
+			return err
+		}
+		id, err := insert(tx, entry.Path, rev, content)
+		if err != nil {
+			return err
+		}
+		next[entry.Path] = doc{id: id, rev: rev}
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	x.known = next
+	x.mu.Lock()
+	if version == x.pendingVersion {
+		x.pending = nil
+	}
+	x.ready = true
+	x.lastError = nil
+	x.mu.Unlock()
+	return nil
+}
+func (x *Index) Optimize(ctx context.Context) error {
+	x.work.Lock()
+	defer x.work.Unlock()
+	for _, stmt := range []string{`INSERT INTO docs_fts(docs_fts) VALUES ('optimize')`, `PRAGMA wal_checkpoint(TRUNCATE)`, `VACUUM`} {
+		if _, err := x.db.ExecContext(ctx, stmt); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (x *Index) syncPending() error {
+	x.work.Lock()
+	defer x.work.Unlock()
+	x.mu.Lock()
+	files := x.pending
+	x.pending = nil
+	x.mu.Unlock()
+	if files == nil {
+		return nil
+	}
+	return x.syncLocked(files)
+}

internal/search/maintenance_test.go

@@ -0,0 +1,56 @@
+package search
+
+import (
+	"context"
+	"testing"
+)
+
+func TestRebuildRepairsUnchangedRevisionAndOptimizePreservesSearch(t *testing.T) {
+	f := newFixture(t)
+	f.write("a.md", "# 中文检索\n![](assets/a.png)")
+	f.sync()
+	if _, err := f.idx.db.Exec(`UPDATE docs SET metadata='{}'`); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.idx.Rebuild(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	notes, ready, err := f.idx.Metadata()
+	if err != nil || !ready || len(notes) != 1 || len(notes[0].Assets) != 1 {
+		t.Fatalf("重建未重新解析: %+v %v", notes, err)
+	}
+	if err = f.idx.Optimize(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	results, err := f.idx.Search("中文", 20)
+	if err != nil || len(results) != 1 {
+		t.Fatalf("优化影响搜索: %+v %v", results, err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err = f.idx.Rebuild(ctx); err == nil {
+		t.Fatal("取消必须报告失败")
+	}
+	results, err = f.idx.Search("中文", 20)
+	if err != nil || len(results) != 1 {
+		t.Fatal("失败重建破坏原索引")
+	}
+}
+
+func TestRebuildSupersedesQueuedOldSnapshot(t *testing.T) {
+	f := newFixture(t)
+	f.write("old.md", "old")
+	f.sync()
+	f.idx.Update(map[string]string{}) // 尚未消费的旧扫描不应在手动重建后删除新索引。
+	f.write("new.md", "中文新笔记")
+	if err := f.idx.Rebuild(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.idx.syncPending(); err != nil {
+		t.Fatal(err)
+	}
+	results, err := f.idx.Search("中文", 20)
+	if err != nil || len(results) != 1 {
+		t.Fatalf("旧快照覆盖重建: %+v %v", results, err)
+	}
+}

internal/search/search.go

@@ -61,7 +61,9 @@ type Index struct {
 	mu             sync.Mutex
 	pending        map[string]string // 待同步的最新快照（路径 -> rev）
 	wake           chan struct{}
+	pendingVersion uint64
 
+	work  sync.Mutex     // 串行化同步、重建和数据库维护
 	known map[string]doc // 已索引的笔记，仅由同步协程访问
 }
 
@@ -113,6 +115,7 @@ func (x *Index) Close() error { return x.db.Close() }
 func (x *Index) Update(files map[string]string) {
 	x.mu.Lock()
 	x.pending = files
+	x.pendingVersion++
 	x.mu.Unlock()
 	select {
 	case x.wake <- struct{}{}:
@@ -127,10 +130,7 @@ func (x *Index) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-x.wake:
-			x.mu.Lock()
-			files := x.pending
-			x.mu.Unlock()
-			if err := x.sync(files); err != nil {
+			if err := x.syncPending(); err != nil {
 				x.mu.Lock()
 				x.lastError = err
 				x.mu.Unlock()
@@ -144,6 +144,12 @@ func (x *Index) Run(ctx context.Context) {
 
 // sync 让索引与快照一致：只读取新增或 rev 变化的笔记
 func (x *Index) sync(files map[string]string) error {
+	x.work.Lock()
+	defer x.work.Unlock()
+	return x.syncLocked(files)
+}
+
+func (x *Index) syncLocked(files map[string]string) error {
 	tx, err := x.db.Begin()
 	if err != nil {
 		return err
@@ -178,21 +184,10 @@ func (x *Index) sync(files map[string]string) error {
 				return err
 			}
 		}
-		metadata, err := json.Marshal(vault.Analyze(p, content))
+		id, err := insert(tx, p, rev, content)
 		if err != nil {
 			return err
 		}
-		res, err := tx.Exec(`INSERT INTO docs (path, rev, metadata) VALUES (?, ?, ?)`, p, rev, string(metadata))
-		if err != nil {
-			return err
-		}
-		id, err := res.LastInsertId()
-		if err != nil {
-			return err
-		}
-		if _, err := tx.Exec(`INSERT INTO docs_fts (rowid, path, body) VALUES (?, ?, ?)`, id, p, string(content)); err != nil {
-			return err
-		}
 		next[p] = doc{id: id, rev: rev}
 	}
 	if err := tx.Commit(); err != nil {