diff --git a/internal/search/maintenance.go b/internal/search/maintenance.go new file mode 100644 index 0000000..a6c58cc --- /dev/null +++ b/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) +} diff --git a/internal/search/maintenance_test.go b/internal/search/maintenance_test.go new file mode 100644 index 0000000..1db1f76 --- /dev/null +++ b/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) + } +} diff --git a/internal/search/search.go b/internal/search/search.go index dccf34d..90eec67 100644 --- a/internal/search/search.go +++ b/internal/search/search.go @@ -58,10 +58,12 @@ type Index struct { db *sql.DB v *vault.Vault - mu sync.Mutex - pending map[string]string // 待同步的最新快照（路径 -> rev） - wake chan 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 {