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
// "<page id>#<date>-<n>" 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
}