package search import ( "context" "errors" "fmt" "os" "path/filepath" "sync" "time" "github.com/blevesearch/bleve/v2" "sourcecraft.dev/bigbes/sr-ht-spec/core" ) // Document is one unit of the index: a whole document, or one dated entry of an // activity log. Extract produces these from a doc.Archive; the index stores // them and hands back Hits shaped the same way. type Document struct { // Space is what makes the one global index filterable per project. Every // document carries it, and a project query is a term filter over the set. Space core.SpaceRef // ID is the document's id within its space: doc.Page.ID, or // "#-" for one entry of an activity log. ID string // Rev is the revision the document was read at, carried through so a hit // can be turned into a pinned `?rev=` URL rather than a link to whatever // the branch says now. Rev string // Path is the document's path in the git tree. For a log entry it is the // path of the log document the entry came out of. Path string // Anchor is the heading anchor within Path a hit should land on. Empty for // an ordinary document, set for a log entry. Anchor string // Section is the top-level directory the document lives under, or // doc.LogSection for an activity log and its entries. Section string Title string // Text is everything searchable: the frontmatter projected to "key: value" // lines followed by the rendered plain text. It is split by language and // stored in the analyzed fields; it is not stored verbatim under its own // name. Text string } // Index is the one global bleve index, shared by every space and every project. // // One index, not one per project and not one per space. Per-project indexes // were specified and then retracted in the design for a concrete reason: with N // projects every merge fans out to N rebuilds and adding a space to a project // forces one, while the "everything" project is a second full copy of the // corpus. Here a project is a filter — see Query.Spaces — so a merge touches // one index and every project containing the space sees the change for free. // // bleve is single-writer. One process holds the index open, which is why the // push hooks are RPC shims into the daemon rather than separate processes. // Within that process an Index is safe for concurrent use: searches share a // read lock, and a rebuild takes the write lock. type Index struct { path string mu sync.RWMutex idx bleve.Index } // batchSize is how many documents are buffered before a batch is flushed // during a full rebuild. Carried over from warren. const batchSize = 200 // Stats reports what a rebuild did and how long it took. // // The design absorbs warren's batch full rebuild deliberately — at tens of // documents a day, incremental indexing is machinery bought against a cost // nobody has measured — and asks for the duration to be instrumented so the // decision to revisit is triggered by a number rather than a hunch. That is // what Took is for: it is the trigger, and callers are expected to log it. type Stats struct { // Spaces is how many distinct spaces the rebuild wrote. Spaces int // Indexed is how many documents were written. Indexed int // Deleted is how many stale documents were removed: documents that were in // the index for a rebuilt space and are not in the new document set. Deleted int Took time.Duration } func (s Stats) String() string { return fmt.Sprintf("spaces=%d indexed=%d deleted=%d took=%s", s.Spaces, s.Indexed, s.Deleted, s.Took.Round(time.Millisecond)) } // Open opens the global index at path, creating an empty one if it is not // there. The index is a pure cache: deleting the directory and letting Open // recreate it, followed by RebuildAll, is always a valid repair. func Open(path string) (*Index, error) { if path == "" { return nil, errors.New("search: index path is required") } idx, err := bleve.Open(path) switch { case errors.Is(err, bleve.ErrorIndexPathDoesNotExist): idx, err = bleve.New(path, buildMapping()) if err != nil { return nil, fmt.Errorf("search: create index at %s: %w", path, err) } case err != nil: return nil, fmt.Errorf("search: open index at %s: %w", path, err) } return &Index{path: path, idx: idx}, nil } // Path is where the index lives on disk. func (x *Index) Path() string { return x.path } // Close releases the index. func (x *Index) Close() error { x.mu.Lock() defer x.mu.Unlock() if x.idx == nil { return nil } err := x.idx.Close() x.idx = nil if err != nil { return fmt.Errorf("search: close index: %w", err) } return nil } // Count is how many documents the index holds, across every space. func (x *Index) Count() (uint64, error) { x.mu.RLock() defer x.mu.RUnlock() if x.idx == nil { return 0, errors.New("search: index is closed") } n, err := x.idx.DocCount() if err != nil { return 0, fmt.Errorf("search: count documents: %w", err) } return n, nil } // RebuildSpace replaces everything the index holds for one space. // // This is a rebuild, not an incremental update: whatever was indexed for sp is // removed and docs are written in its place, with no diffing of individual // documents and no per-document staleness bookkeeping. The unit of freshness is // a space at a revision, which is exactly what the index_stamp row in Postgres // records — this package is handed a revision's worth of documents and does not // know or care which of them changed. // // docs may be empty, which empties the space. Every document must belong to sp; // a document from another space is a caller bug and is refused rather than // written somewhere surprising. func (x *Index) RebuildSpace(ctx context.Context, sp core.SpaceRef, docs []Document) (Stats, error) { start := time.Now() if sp.Owner == "" || sp.Name == "" { return Stats{}, errors.New("search: RebuildSpace needs a space") } fields := make([]map[string]any, len(docs)) for i, d := range docs { if d.Space != sp { return Stats{}, fmt.Errorf("search: document %q belongs to space %s, not %s", d.ID, d.Space, sp) } f, err := bleveDoc(d) if err != nil { return Stats{}, err } fields[i] = f } x.mu.Lock() defer x.mu.Unlock() if x.idx == nil { return Stats{}, errors.New("search: index is closed") } stale, err := x.keysOf(sp) if err != nil { return Stats{}, err } if err := ctx.Err(); err != nil { return Stats{}, err } // One batch for the whole space, so a space is never half-replaced. Deletes // go in first: a batch is keyed by document id and the last operation on a // key wins, so a document that survives the rebuild is re-indexed rather // than dropped. batch := x.idx.NewBatch() kept := 0 for key := range stale { batch.Delete(key) } for i, d := range docs { key := Key(d.Space, d.ID) if _, ok := stale[key]; ok { kept++ } if err := batch.Index(key, fields[i]); err != nil { return Stats{}, fmt.Errorf("search: stage %s: %w", key, err) } } if err := x.idx.Batch(batch); err != nil { return Stats{}, fmt.Errorf("search: rebuild space %s: %w", sp, err) } st := Stats{Spaces: 1, Indexed: len(docs), Deleted: len(stale) - kept, Took: time.Since(start)} if len(docs) == 0 { st.Spaces = 0 } return st, nil } // DeleteSpace removes every document of a space from the index. It is what a // deleted space calls; RebuildSpace with no documents does the same thing. func (x *Index) DeleteSpace(ctx context.Context, sp core.SpaceRef) (Stats, error) { return x.RebuildSpace(ctx, sp, nil) } // RebuildAll replaces the entire index with docs, which may span any number of // spaces. // // The rebuild runs into a fresh index beside the live one and the two are // swapped at the end, so a failure part-way through leaves the old index intact // and serving. That is the one thing warren's `bleve.New` over the live path // did not give, and it matters here because the index is open in a daemon that // is answering queries while the rebuild runs. func (x *Index) RebuildAll(ctx context.Context, docs []Document) (Stats, error) { start := time.Now() spaces := make(map[core.SpaceRef]struct{}) fields := make([]map[string]any, len(docs)) for i, d := range docs { f, err := bleveDoc(d) if err != nil { return Stats{}, err } fields[i] = f spaces[d.Space] = struct{}{} } tmp := x.path + ".rebuilding" if err := os.RemoveAll(tmp); err != nil { return Stats{}, fmt.Errorf("search: clear %s: %w", tmp, err) } fresh, err := bleve.New(tmp, buildMapping()) if err != nil { return Stats{}, fmt.Errorf("search: create index at %s: %w", tmp, err) } if err := indexAll(ctx, fresh, docs, fields); err != nil { _ = fresh.Close() _ = os.RemoveAll(tmp) return Stats{}, err } if err := fresh.Close(); err != nil { _ = os.RemoveAll(tmp) return Stats{}, fmt.Errorf("search: close rebuilt index: %w", err) } x.mu.Lock() defer x.mu.Unlock() if x.idx == nil { _ = os.RemoveAll(tmp) return Stats{}, errors.New("search: index is closed") } if err := x.swap(tmp); err != nil { return Stats{}, err } return Stats{ Spaces: len(spaces), Indexed: len(docs), Took: time.Since(start), }, nil } // swap puts the freshly built index at tmp in place of the live one. The caller // holds the write lock. // // The old directory is renamed aside rather than deleted first, so the window in // which neither exists is a rename rather than a recursive delete. If reopening // the new index fails the Index is left closed and the error is returned: the // index is a cache, and a caller that cannot open it must rebuild it, not // silently serve an empty one. func (x *Index) swap(tmp string) error { if err := x.idx.Close(); err != nil { x.idx = nil return fmt.Errorf("search: close live index: %w", err) } x.idx = nil old := x.path + ".old" if err := os.RemoveAll(old); err != nil { return fmt.Errorf("search: clear %s: %w", old, err) } if err := os.Rename(x.path, old); err != nil && !errors.Is(err, os.ErrNotExist) { return fmt.Errorf("search: move live index aside: %w", err) } if err := os.Rename(tmp, x.path); err != nil { return fmt.Errorf("search: move rebuilt index into place: %w", err) } idx, err := bleve.Open(x.path) if err != nil { return fmt.Errorf("search: reopen index at %s: %w", x.path, err) } x.idx = idx if err := os.RemoveAll(old); err != nil { return fmt.Errorf("search: remove %s: %w", old, err) } return nil } func indexAll(ctx context.Context, idx bleve.Index, docs []Document, fields []map[string]any) error { batch := idx.NewBatch() for i, d := range docs { if err := ctx.Err(); err != nil { return err } key := Key(d.Space, d.ID) if err := batch.Index(key, fields[i]); err != nil { return fmt.Errorf("search: stage %s: %w", key, err) } if batch.Size() >= batchSize { if err := idx.Batch(batch); err != nil { return fmt.Errorf("search: flush batch: %w", err) } batch = idx.NewBatch() } } if batch.Size() > 0 { if err := idx.Batch(batch); err != nil { return fmt.Errorf("search: flush final batch: %w", err) } } return nil } // keysOf lists the index keys currently held for a space. The caller holds a // lock. func (x *Index) keysOf(sp core.SpaceRef) (map[string]struct{}, error) { q := bleve.NewTermQuery(sp.String()) q.SetField(fieldSpace) const page = 1000 keys := make(map[string]struct{}) for from := 0; ; from += page { req := bleve.NewSearchRequestOptions(q, page, from, false) res, err := x.idx.Search(req) if err != nil { return nil, fmt.Errorf("search: list documents of %s: %w", sp, err) } for _, h := range res.Hits { keys[h.ID] = struct{}{} } if len(res.Hits) < page { return keys, nil } } } // tmpPaths are the working directories a rebuild uses, exposed only so a caller // cleaning up after a crash knows what to look for. func tmpPaths(path string) []string { return []string{path + ".rebuilding", path + ".old"} } // CleanStale removes the working directories a crashed rebuild may have left // behind next to the index. Safe to call at startup, before Open. func CleanStale(path string) error { for _, p := range tmpPaths(path) { if err := os.RemoveAll(p); err != nil { return fmt.Errorf("search: remove %s: %w", filepath.Base(p), err) } } return nil }