mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/tools.go

package main

// The module's tools (novox/hq to-be 55 §1.1, the instruction tools-are-how-a-module-is-spoken-to): the one way a person or a
// session reads and steers what the module keeps. Every state has a tool that reads it, every act the module does by itself
// can be done by hand, and every recovery has a tool.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"

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

// The module's tools beside the seat's verbs.
const (
	HeadTool       = "knowledge_head"
	OperationsTool = "knowledge_operations"
	SectionTool    = "knowledge_section"
	DraftTool      = "knowledge_draft"
	PartsTool      = "knowledge_parts"
	AskedTool      = "knowledge_asked"
	FillTool       = "knowledge_fill"
	IndexTool      = "knowledge_rebuild_index"
	AgainTool      = "knowledge_review_again"
	IntegrityTool  = "knowledge_integrity"
	MoveTool       = "knowledge_move"
	MoveCheckTool  = "knowledge_move_check"
)

// ModuleTools are the module's own tools beside the health tool, in the order the manifest lists them.
var ModuleTools = []string{HealthTool, HighestTool, MarksTool, HeadTool, OperationsTool, SectionTool, DraftTool, PartsTool,
	AskedTool, FillTool, IndexTool, AgainTool, IntegrityTool, MoveTool, MoveCheckTool}

var errNotOnLedger = errors.New("the records are still in the database: this tool reads the ledger, which serves them after the move")

// ledger is the ledger store the records are served from, or the refusal that they are not.
func (k *KB) ledger() (*ledgerStore, error) {
	switch s := k.Store.(type) {
	case *ledgerStore:
		return s, nil
	case *movableStore:
		if s.OnLedger() {
			return s.led, nil
		}
	}
	return nil, errNotOnLedger
}

func (k *KB) movable() (*movableStore, error) {
	if m, ok := k.Store.(*movableStore); ok {
		return m, nil
	}
	return nil, errors.New("this knowledge base has no database to move from")
}

func keyArgs(name string, m map[string]any, more ...string) (Key, args, error) {
	a, err := readArgs(name, m, append([]string{"kind", "number"}, more...)...)
	if err != nil {
		return Key{}, args{}, err
	}
	k, err := a.key()
	return k, a, err
}

func keyProps(more map[string]any) map[string]any {
	p := map[string]any{"kind": kindArg(), "number": integer("the record's number")}
	for k, v := range more {
		p[k] = v
	}
	return p
}

// moduleTools are the tools beside the seat's verbs that read and steer the records on the bus.
func moduleTools(k *KB, caller func(map[string]any) string) []stdio.Tool {
	return []stdio.Tool{
		{Name: HeadTool,
			Description: "One record's head in the ledger, whole: its fields, clauses, links, its served and proposed versions, its sections as " +
				"title, digest and the operation that wrote each, and the sequence of the last operation it holds. Refused while the records are in the database.",
			Input: schema([]string{"number"}, keyProps(nil)),
			Run: func(m map[string]any) (any, error) {
				key, _, err := keyArgs(HeadTool, m)
				if err != nil {
					return nil, err
				}
				l, err := k.ledger()
				if err != nil {
					return nil, err
				}
				l.mu.Lock()
				defer l.mu.Unlock()
				h, err := l.fresh(key)
				if err != nil || h == nil {
					return nil, fmt.Errorf("%s is held of none (%v)", key, err)
				}
				return h, nil
			}},
		{Name: OperationsTool,
			Description: "A record's history in the ledger, oldest first from a sequence: each operation with its sequence, its name, who it was by " +
				"(the module's account and the caller the runtime proved) and when. A section's text is not shown; knowledge_section reads it.",
			Input: schema([]string{"number"}, keyProps(map[string]any{"from": integer("the sequence to read from"), "limit": integer("at most this many")})),
			Run: func(m map[string]any) (any, error) {
				key, a, err := keyArgs(OperationsTool, m, "from", "limit")
				if err != nil {
					return nil, err
				}
				from, _ := a.num("from", false)
				limit, _ := a.num("limit", false)
				l, err := k.ledger()
				if err != nil {
					return nil, err
				}
				var page buscalls.HistoryPage
				if err := l.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(key), From: uint64(from), Limit: limit}, &page); err != nil {
					return nil, err
				}
				for i := range page.Operations {
					page.Operations[i].Body = nil // a section's text and a head are read by their own tools
				}
				return page, nil
			}},
		{Name: SectionTool,
			Description: "One section's text, by the sequence of the write-section operation that wrote it, checked against its digest. Refused when the " +
				"operation is no write-section of the record, or its text does not hash to its digest (the history is damaged).",
			Input: schema([]string{"number", "sequence"}, keyProps(map[string]any{"sequence": integer("the sequence of the write-section")})),
			Run: func(m map[string]any) (any, error) {
				key, a, err := keyArgs(SectionTool, m, "sequence")
				if err != nil {
					return nil, err
				}
				seq, err := a.num("sequence", true)
				if err != nil {
					return nil, err
				}
				l, err := k.ledger()
				if err != nil {
					return nil, err
				}
				var page buscalls.HistoryPage
				if err := l.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(key), From: uint64(seq), Limit: 1}, &page); err != nil {
					return nil, err
				}
				if len(page.Operations) == 0 || page.Operations[0].Sequence != uint64(seq) || page.Operations[0].Operation != OpWriteSection {
					return nil, fmt.Errorf("operation %d of %s is no write-section", seq, key)
				}
				var b struct{ Title, Text, Digest string }
				if err := json.Unmarshal(page.Operations[0].Body, &b); err != nil {
					return nil, err
				}
				if SectionDigest(b.Text) != b.Digest {
					return nil, fmt.Errorf("the text of operation %d does not hash to its digest %s: the history is damaged", seq, b.Digest)
				}
				return map[string]any{"title": b.Title, "digest": b.Digest, "text": b.Text, "author": page.Operations[0].Author, "at": page.Operations[0].At}, nil
			}},
		{Name: DraftTool,
			Description: "A record's draft: its author, since when, the version it starts from, its fields and its sections as title, digest and the operation that wrote each. " +
				"Answers that there is none when none is open.",
			Input: schema([]string{"number"}, keyProps(nil)),
			Run: func(m map[string]any) (any, error) {
				key, _, err := keyArgs(DraftTool, m)
				if err != nil {
					return nil, err
				}
				d, _, err := k.readDraft(key)
				if err != nil {
					return nil, err
				}
				if d == nil {
					return map[string]any{"record": key.String(), "says": "no draft is open"}, nil
				}
				return d, nil
			}},
		{Name: PartsTool,
			Description: "The latest verdict of each part of a record's version being worked on or served: the head and each section, the digest it was judged with, " +
				"its verdict and findings, who judged and when.",
			Input: schema([]string{"number"}, keyProps(nil)),
			Run: func(m map[string]any) (any, error) {
				key, _, err := keyArgs(PartsTool, m)
				if err != nil {
					return nil, err
				}
				e, err := k.Store.Get(context.Background(), key)
				if err != nil || e == nil {
					return nil, fmt.Errorf("%s is held of none (%v)", key, err)
				}
				return map[string]any{"record": key.String(), "parts": e.Working().Parts}, nil
			}},
		{Name: AskedTool,
			Description: "The parts asked of the judge and not answered yet: the record, the part and since when. With clear and a key, takes one mark away " +
				"(a part whose answer will not come); the record's review asks it again.",
			Input: schema(nil, map[string]any{"clear": text("a key to take away")}),
			Run: func(m map[string]any) (any, error) {
				a, err := readArgs(AskedTool, m, "clear")
				if err != nil {
					return nil, err
				}
				kv := k.States.State(askedState)
				if c, _ := a.str("clear"); c != "" {
					return map[string]any{"cleared": c}, kv.Delete(c)
				}
				keys, err := kv.Keys()
				if err != nil {
					return nil, err
				}
				sort.Strings(keys)
				out := []any{}
				for _, key := range keys {
					if e, err := kv.Get(key); err == nil && e != nil {
						var v any
						_ = json.Unmarshal(e.Value, &v)
						out = append(out, map[string]any{"key": key, "asked": v})
					}
				}
				return map[string]any{"asked": out}, nil
			}},
		{Name: FillTool,
			Description: "How full the module's kept data is against its caps: the stream of the parts' verdicts (messages, bytes and its cap), and the number of keys of each state.",
			Input:       schema(nil, map[string]any{}),
			Run: func(m map[string]any) (any, error) {
				if _, err := readArgs(FillTool, m); err != nil {
					return nil, err
				}
				out := map[string]any{}
				if l, err := k.ledger(); err == nil {
					var f buscalls.StreamFilled
					if err := l.pipe.Call(buscalls.StreamFill, buscalls.StreamRequest{Stream: "reviews"}, &f); err != nil {
						out["reviews"] = "cannot be read: " + err.Error()
					} else {
						out["reviews"] = f
					}
				}
				keys := map[string]int{}
				for _, s := range []string{draftsState, askedState, findingsState, sentState} {
					if ks, err := k.States.State(s).Keys(); err == nil {
						keys[s] = len(ks)
					}
				}
				out["keys"] = keys
				return out, nil
			}},
		{Name: IndexTool,
			Description: "Build the search index again from the ledger now, and say how it stands: built, when, and how many documents it holds. ask and holds answer " +
				"\"index not built\" meanwhile.",
			Input: schema(nil, map[string]any{}),
			Run: func(m map[string]any) (any, error) {
				if _, err := readArgs(IndexTool, m); err != nil {
					return nil, err
				}
				l, err := k.ledger()
				if err != nil {
					return nil, err
				}
				if err := l.BuildIndex(context.Background()); err != nil {
					return nil, err
				}
				return l.IndexState(), nil
			}},
		{Name: AgainTool,
			Description: "Ask a record's review again, for a version still proposed: its parts that approved keep their verdicts and the others are judged " +
				"again; the review that ran out of tries starts its tries anew. Refused for a record that is not proposed.",
			Input: schema([]string{"number"}, keyProps(nil)),
			Run: func(m map[string]any) (any, error) {
				key, _, err := keyArgs(AgainTool, m)
				if err != nil {
					return nil, err
				}
				return k.ReviewAgain(caller(m), key)
			}},
		{Name: IntegrityTool,
			Description: "Check that every section a head or a draft names resolves to a write-section whose text hashes to its digest, for one record or for " +
				"all of them, and say each that does not.",
			Input: schema(nil, keyProps(nil)),
			Run: func(m map[string]any) (any, error) {
				a, err := readArgs(IntegrityTool, m, "kind", "number")
				if err != nil {
					return nil, err
				}
				l, err := k.ledger()
				if err != nil {
					return nil, err
				}
				var keys []Key
				if _, has := m["number"]; has {
					key, err := a.key()
					if err != nil {
						return nil, err
					}
					keys = []Key{key}
				} else {
					all, err := l.All(context.Background())
					if err != nil {
						return nil, err
					}
					for _, e := range all {
						keys = append(keys, e.Current.Key())
					}
				}
				bad := []string{}
				l.mu.Lock()
				defer l.mu.Unlock()
				for _, key := range keys {
					h, err := l.fresh(key)
					if err != nil || h == nil {
						bad = append(bad, fmt.Sprintf("%s: %v", key, err))
						continue
					}
					for _, sr := range []*storedRecord{h.Entry.Current, h.Entry.Pending} {
						if sr == nil {
							continue
						}
						for _, ref := range sr.Sections {
							delete(l.texts, ref.Digest) // read from the history, not from what was held
							if _, err := l.text(ref, key); err != nil {
								bad = append(bad, fmt.Sprintf("%s: %v", key, err))
							}
						}
					}
				}
				return map[string]any{"checked": len(keys), "damaged": bad}, nil
			}},
		{Name: MoveTool,
			Description: "Move the records from the database to the ledger: with dry-run, say what a copy would write and write nothing; with commit, refuse writes for " +
				"its length, copy every version of every record, check the copy record by record, and set the records to be served from the ledger only if the check is " +
				"clean. Run again after a copy it changes nothing. Refused once the records are served from the ledger.",
			Input: schema(nil, map[string]any{"dry-run": flag("count and write nothing"), "commit": flag("copy, check and switch")}),
			Run: func(m map[string]any) (any, error) {
				a, err := readArgs(MoveTool, m, "dry-run", "commit")
				if err != nil {
					return nil, err
				}
				var ask MoveAsk
				if ask.DryRun, err = a.flag("dry-run"); err != nil {
					return nil, err
				}
				if ask.Commit, err = a.flag("commit"); err != nil {
					return nil, err
				}
				ms, err := k.movable()
				if err != nil {
					return nil, err
				}
				return ms.Move(context.Background(), ask, func(s string) { k.logf("[mesh-knowledge] the move: %s", s) })
			}},
		{Name: MoveCheckTool,
			Description: "Compare the database with the ledger record by record now: the versions, the served and the proposed version whole and every section's text. " +
				"Answers every difference, at most 50, and where the records are served from; a database that is behind is named.",
			Input: schema(nil, map[string]any{}),
			Run: func(m map[string]any) (any, error) {
				if _, err := readArgs(MoveCheckTool, m); err != nil {
					return nil, err
				}
				ms, err := k.movable()
				if err != nil {
					return nil, err
				}
				diffs, err := ms.Check(context.Background())
				if err != nil {
					return nil, err
				}
				return map[string]any{"source": ms.Source(), "differences": diffs, "clean": len(diffs) == 0}, nil
			}},
	}
}

// Review is the record's review asked again.
type ReviewedAgain struct {
	Record string `json:"record"`
	Says   string `json:"says"`
}

// ReviewAgain asks the review of a proposed version again: a failed review starts its tries anew, and a review whose verdict
// was changes is made due, the parts that approved keeping their verdicts.
func (k *KB) ReviewAgain(caller string, key Key) (*ReviewedAgain, error) {
	k.mu.Lock()
	defer k.mu.Unlock()
	ctx := k.ctxFor(caller)
	e, err := k.Store.Get(ctx, key)
	if err != nil || e == nil {
		return nil, fmt.Errorf("%s is held of none (%v)", key, err)
	}
	w := e.Working().Clone()
	if w.Status != StatusProposed {
		return nil, fmt.Errorf("%s is %s, not proposed: there is no review to ask again", key, w.Status)
	}
	if !proven(caller) {
		return nil, fmt.Errorf("review-again needs a caller the runtime proved; this call is by %s", caller)
	}
	w.Judging = &Judging{State: JudgingDue}
	if err := k.Store.Save(ctx, Version{By: caller, At: k.now(), Why: "its review is asked again", Record: w, Insight: e.Pending != nil && e.Pending.IsInsight()}, e.Pending == nil, false, e.Revision); err != nil {
		return nil, err
	}
	k.wake()
	return &ReviewedAgain{Record: key.String(), Says: "The review of " + key.String() + " is due again; the parts that approved keep their verdicts."}, nil
}