mesh-catalog/modules/mesh-issues/cmd/mesh-issues/view.go

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"sync"
	"unicode"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// memoryView is every folded issue in memory, kept current by watching the module's own bucket, with a full-text
// index of title, report and notes (novox/hq ADR 0300 §5). Until it is loaded on start, the tracker reads
// the bucket as before.
type memoryView struct {
	mu     sync.RWMutex
	loaded bool
	issues map[uint64]*Issue
	index  *index
}

func newMemoryView() *memoryView { return &memoryView{issues: map[uint64]*Issue{}, index: newIndex()} }

// view is the tracker's issues in memory.
func (t *Tracker) view() *memoryView {
	t.heldOnce.Do(func() { t.held = newMemoryView() })
	return t.held
}

// load holds these issues and rebuilds the index from them. An issue the watch handed over meanwhile is kept
// where it was folded later than the one read.
func (h *memoryView) load(all map[uint64]*Issue) {
	h.mu.Lock()
	defer h.mu.Unlock()
	merged := map[uint64]*Issue{}
	for n, is := range all {
		merged[n] = is
	}
	for n, is := range h.issues {
		if old := merged[n]; old == nil || old.Folded < is.Folded {
			merged[n] = is
		}
	}
	h.issues = merged
	h.index = newIndex()
	for _, is := range merged {
		h.index.add(is)
	}
	h.loaded = true
}

func (h *memoryView) ready() bool {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.loaded
}

// put holds an issue unless what is held was folded later. Before the load it is held for the load to merge.
func (h *memoryView) put(is *Issue) {
	h.mu.Lock()
	defer h.mu.Unlock()
	old := h.issues[is.Number]
	if old != nil && old.Folded > is.Folded {
		return
	}
	h.issues[is.Number] = is
	if h.loaded {
		if old != nil {
			h.index.remove(old.Number)
		}
		h.index.add(is)
	}
}

func (h *memoryView) drop(n uint64) {
	h.mu.Lock()
	defer h.mu.Unlock()
	delete(h.issues, n)
	if h.loaded {
		h.index.remove(n)
	}
}

// all is every issue held, each a shallow copy: a caller may set an issue's fields without changing what is held,
// but its lists (notes, links, fixes and the rest) are shared with what is held, so a caller reads them and never
// changes them in place. Every caller in this module only reads; a write goes through change, which reads the
// bucket afresh. A deep copy per list would cost a JSON round trip of every issue on every list.
func (h *memoryView) all() map[uint64]*Issue {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make(map[uint64]*Issue, len(h.issues))
	for n, is := range h.issues {
		c := *is
		out[n] = &c
	}
	return out
}

func (h *memoryView) search(q string) []hit {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.index.search(q)
}

// remember holds an issue the tracker just wrote, so a list right after a write shows it before the watch
// hands it back.
func (t *Tracker) remember(is *Issue) {
	if is == nil {
		return
	}
	c, err := clone(is)
	if err != nil {
		return
	}
	t.view().put(c)
}

// Watched takes one change of the module's own bucket into memory: a put held unless memory holds a later
// fold, a delete dropped (the next start folds it back from the log).
func (t *Tracker) Watched(c stdio.StateChange) error {
	n, ok := numberOfKey(c.Key)
	if !ok {
		return nil
	}
	if c.Op != "put" {
		t.view().drop(n)
		return nil
	}
	var is Issue
	if err := json.Unmarshal(c.Value, &is); err != nil {
		return fmt.Errorf("issue %d is kept in a form this module cannot read: %v", n, err)
	}
	t.view().put(&is)
	return nil
}

// readAll is every issue the bucket holds, by number, each read by its key.
func (t *Tracker) readAll() (map[uint64]*Issue, error) {
	keys, err := t.Store.Keys(IssuesState)
	if err != nil {
		return nil, fmt.Errorf("the issues cannot be listed: %v", err)
	}
	out := map[uint64]*Issue{}
	for _, k := range keys {
		n, ok := numberOfKey(k)
		if !ok {
			continue
		}
		is, _, err := read(t.Store, n)
		if errors.Is(err, ErrNoIssue) {
			continue
		}
		if err != nil {
			return nil, err
		}
		out[n] = is
	}
	return out, nil
}

// allIssues is every issue kept, by number: from memory once it is loaded, from the bucket before.
func (t *Tracker) allIssues() (map[uint64]*Issue, error) {
	if v := t.view(); v.ready() {
		return v.all(), nil
	}
	return t.readAll()
}

// index is a full-text index of the issues' title, report and notes, ranked by BM25. Written here rather than
// taken from a library (bleve): the tracker holds a few thousand issues, the index is rebuilt on every start in
// well under a second, and its whole need — words in, issues ranked out — is a hundred lines, against a
// dependency of dozens of modules for a bundle that has two.
type index struct {
	// postings is, for each word, how often it stands in each issue, the title's words counted titleWeight
	// times.
	postings map[string]map[uint64]int
	// lengths is each issue's words, weighted as postings counts them; total their sum.
	lengths map[uint64]int
	words   map[uint64][]string
	total   int
}

const titleWeight = 3

func newIndex() *index {
	return &index{postings: map[string]map[uint64]int{}, lengths: map[uint64]int{}, words: map[uint64][]string{}}
}

// tokens is a text's words: lower case, letters and digits, two characters at least.
func tokens(s string) []string {
	var out []string
	for _, w := range strings.FieldsFunc(strings.ToLower(s), func(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsDigit(r) }) {
		if len([]rune(w)) >= 2 {
			out = append(out, w)
		}
	}
	return out
}

func (x *index) add(is *Issue) {
	counts := map[string]int{}
	for _, w := range tokens(is.Title) {
		counts[w] += titleWeight
	}
	body := []string{is.Report}
	for _, n := range is.Notes {
		body = append(body, n.Text)
	}
	for _, w := range tokens(strings.Join(body, "\n")) {
		counts[w]++
	}
	length := 0
	words := make([]string, 0, len(counts))
	for w, c := range counts {
		p := x.postings[w]
		if p == nil {
			p = map[uint64]int{}
			x.postings[w] = p
		}
		p[is.Number] = c
		length += c
		words = append(words, w)
	}
	x.lengths[is.Number] = length
	x.words[is.Number] = words
	x.total += length
}

func (x *index) remove(n uint64) {
	for _, w := range x.words[n] {
		if p := x.postings[w]; p != nil {
			delete(p, n)
			if len(p) == 0 {
				delete(x.postings, w)
			}
		}
	}
	x.total -= x.lengths[n]
	delete(x.lengths, n)
	delete(x.words, n)
}

// hit is one issue a search found, and how well it matched.
type hit struct {
	Number uint64
	Score  float64
}

// search ranks every issue holding at least one of the query's words by BM25: a word few issues hold weighs
// more, a word standing often weighs more with diminishing return, a long issue weighs each word less. Equal
// scores rank the newer first.
func (x *index) search(q string) []hit {
	docs := len(x.lengths)
	if docs == 0 {
		return nil
	}
	const k1, b = 1.2, 0.75
	avg := float64(x.total) / float64(docs)
	scores := map[uint64]float64{}
	seen := map[string]bool{}
	for _, w := range tokens(q) {
		if seen[w] {
			continue
		}
		seen[w] = true
		p := x.postings[w]
		if len(p) == 0 {
			continue
		}
		idf := math.Log(1 + (float64(docs)-float64(len(p))+0.5)/(float64(len(p))+0.5))
		for n, tf := range p {
			f := float64(tf)
			scores[n] += idf * f * (k1 + 1) / (f + k1*(1-b+b*float64(x.lengths[n])/avg))
		}
	}
	out := make([]hit, 0, len(scores))
	for n, s := range scores {
		out = append(out, hit{n, s})
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Score != out[j].Score {
			return out[i].Score > out[j].Score
		}
		return out[i].Number > out[j].Number
	})
	return out
}