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

package main

// The move from the database to the ledger (novox/hq to-be 55 §8.1, ADR 0592 §13 to §15, ADR 0597 §13): a dry run says what
// would be copied; the copy writes every version of every record as an operation (the old author kept apart, marked
// imported), with writes refused for its length (the window); a check compares the two record by record; the switch key is
// set only on a clean check; from then on the ledger is the source and the database is written too until it is dropped
// on the operator's yes, behind or not, and the check runs at every start before the drop.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
)

// The two places the records are served from.
const (
	SourceDatabase = "database"
	SourceLedger   = "ledger"
)

const (
	switchKey   = "switch"
	sequenceKey = "database.sequence"
	behindKey   = "database.behind"
	checkKey    = "move.check"
)

// ErrMoving is the answer of a write made while the records are copied.
var ErrMoving = errors.New("the knowledge base is moving its records to the ledger: writes wait until the copy is done and checked, a few minutes at most")

// movableStore is the Store of a module that is moving: the database until the switch key says the ledger, then the ledger
// with the database written beside it.
type movableStore struct {
	db  Store
	led *ledgerStore
	st  States

	mu     sync.Mutex
	moving bool
	active string
}

// newMovableStore is the store over both places; the switch key is read when it is first used.
func newMovableStore(db Store, led *ledgerStore, st States) *movableStore {
	return &movableStore{db: db, led: led, st: st}
}

func (m *movableStore) kv() KV { return m.st.State(findingsState) }

// Source says where the records are served from, reading the switch key once.
func (m *movableStore) Source() string {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.active == "" {
		m.active = SourceDatabase
		if e, err := m.kv().Get(switchKey); err == nil && e != nil {
			var s string
			if json.Unmarshal(e.Value, &s) == nil && s == SourceLedger {
				m.active = SourceLedger
			}
		}
	}
	return m.active
}

func (m *movableStore) cur() Store {
	if m.Source() == SourceLedger {
		return m.led
	}
	return m.db
}

// OnLedger says the records are served from the ledger: a draft is written section by section only then.
func (m *movableStore) OnLedger() bool { return m.Source() == SourceLedger }

func (m *movableStore) writable() error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.moving {
		return ErrMoving
	}
	return nil
}

func (m *movableStore) Get(ctx context.Context, k Key) (*Entry, error) { return m.cur().Get(ctx, k) }
func (m *movableStore) All(ctx context.Context) ([]*Entry, error)      { return m.cur().All(ctx) }
func (m *movableStore) Versions(ctx context.Context, k Key) ([]Version, error) {
	return m.cur().Versions(ctx, k)
}
func (m *movableStore) Search(ctx context.Context, q string, n int) ([]Hit, error) {
	return m.cur().Search(ctx, q, n)
}
func (m *movableStore) Inbound(ctx context.Context, k Key) ([]Inbound, error) {
	return m.cur().Inbound(ctx, k)
}
func (m *movableStore) SetFindings(ctx context.Context, f []Finding) error {
	return m.cur().SetFindings(ctx, f)
}
func (m *movableStore) Findings(ctx context.Context) ([]Finding, error)  { return m.cur().Findings(ctx) }
func (m *movableStore) Told(ctx context.Context, t string) (bool, error) { return m.cur().Told(ctx, t) }
func (m *movableStore) ToldOf(ctx context.Context, p string) ([]string, error) {
	return m.cur().ToldOf(ctx, p)
}
func (m *movableStore) Untold(ctx context.Context, k string) error { return m.cur().Untold(ctx, k) }

// Create writes to the place served from; once the ledger is, the database is written too (behind or not, ADR 0592 §14).
func (m *movableStore) Create(ctx context.Context, v Version) error {
	if err := m.writable(); err != nil {
		return err
	}
	if m.Source() != SourceLedger {
		return m.db.Create(ctx, v)
	}
	if err := m.led.Create(ctx, v); err != nil {
		return err
	}
	m.mirror(ctx, v.Record.Key(), func() error { return m.db.Create(ctx, v) })
	return nil
}

func (m *movableStore) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error {
	if err := m.writable(); err != nil {
		return err
	}
	if m.Source() != SourceLedger {
		return m.db.Save(ctx, v, current, clearPending, revision)
	}
	if err := m.led.Save(ctx, v, current, clearPending, revision); err != nil {
		return err
	}
	m.mirror(ctx, v.Record.Key(), func() error {
		e, err := m.db.Get(ctx, v.Record.Key())
		if err != nil || e == nil {
			return fmt.Errorf("the database holds no %s: %v", v.Record.Key(), err)
		}
		return m.db.Save(ctx, v, current, clearPending, e.Revision)
	})
	return nil
}

// mirror writes the database beside the ledger. A write the database refuses is not the ledger's refusal: the ledger took
// it, and the database is behind; `database.behind` says which record, and the next check names it.
func (m *movableStore) mirror(ctx context.Context, k Key, write func() error) {
	if err := write(); err != nil {
		_, _ = m.kv().Put(behindKey, map[string]any{"record": k.String(), "error": err.Error(), "at": time.Now().UTC()})
		return
	}
	if e, err := m.kv().Get(sequenceKey); err == nil {
		var n int
		rev := uint64(0)
		if e != nil {
			_ = json.Unmarshal(e.Value, &n)
			rev = e.Revision
		}
		if rev == 0 {
			_, _ = m.kv().Create(sequenceKey, n+1)
		} else {
			_, _ = m.kv().Update(sequenceKey, n+1, rev)
		}
	}
}

// The ledger's own capabilities, offered once the records are served from it.
func (m *movableStore) WriteSection(ctx context.Context, k Key, title, text string) (sectionRef, error) {
	if err := m.writable(); err != nil {
		return sectionRef{}, err
	}
	return m.led.WriteSection(ctx, k, title, text)
}
func (m *movableStore) SectionText(ctx context.Context, k Key, r sectionRef) (string, error) {
	return m.led.SectionText(ctx, k, r)
}
func (m *movableStore) Working(ctx context.Context, k Key) ([]sectionRef, error) {
	return m.led.Working(ctx, k)
}
func (m *movableStore) Known(k Key, refs []sectionRef) { m.led.Known(k, refs) }
func (m *movableStore) Operations(ctx context.Context, k Key) ([]OperationInfo, error) {
	return m.led.Operations(ctx, k)
}

// ---- the copy and its check ------------------------------------------------------------------------------------

// MoveAsk asks the move: a dry run says what would be copied and writes nothing; commit copies, checks and switches.
type MoveAsk struct {
	DryRun bool `json:"dry-run,omitempty"`
	Commit bool `json:"commit,omitempty"`
}

// Moved is the answer to the move.
type Moved struct {
	Records  int      `json:"records"`
	Versions int      `json:"versions"`
	Sections int      `json:"sections"`
	Bytes    int      `json:"bytes"`
	Dry      bool     `json:"dry-run,omitempty"`
	Switched bool     `json:"switched"`
	Source   string   `json:"source"`
	Problems []string `json:"differences,omitempty"`
	Says     string   `json:"says"`
}

// Move copies every record of the database as operations into the ledger. A dry run counts what it would copy. A commit
// refuses writes for its length, copies what the ledger does not hold yet (run again it changes nothing), checks the
// copy and sets the switch key only when the check is clean.
func (m *movableStore) Move(ctx context.Context, a MoveAsk, say func(string)) (*Moved, error) {
	if a.DryRun == a.Commit {
		return nil, errors.New("the move takes dry-run or commit, one of them: a dry run writes nothing and says what it would copy")
	}
	if m.Source() == SourceLedger {
		return nil, errors.New("the records are already served from the ledger; the check says whether the database still matches")
	}
	all, err := m.db.All(ctx)
	if err != nil {
		return nil, fmt.Errorf("the database cannot be read: %v", err)
	}
	out := &Moved{Dry: a.DryRun, Source: SourceDatabase}
	type plan struct {
		e        *Entry
		versions []Version
	}
	var plans []plan
	for _, e := range all {
		vs, err := m.db.Versions(ctx, e.Current.Key())
		if err != nil {
			return nil, err
		}
		plans = append(plans, plan{e, vs})
		out.Records++
		out.Versions += len(vs)
		for _, v := range vs {
			out.Sections += len(v.Record.Sections)
			for _, s := range v.Record.Sections {
				out.Bytes += len(s.Text)
			}
		}
	}
	if a.DryRun {
		out.Says = fmt.Sprintf("A commit would copy %d records, %d versions, %d sections (%d bytes of text) into the ledger, and set the records to be served from it only if the check is clean. Nothing was written.", out.Records, out.Versions, out.Sections, out.Bytes)
		return out, nil
	}
	m.mu.Lock()
	m.moving = true
	m.mu.Unlock()
	defer func() { m.mu.Lock(); m.moving = false; m.mu.Unlock() }()
	copied := 0
	for _, p := range plans {
		k := p.e.Current.Key()
		if held, err := m.led.Get(ctx, k); err != nil {
			return nil, err
		} else if held != nil && held.Versions == len(p.versions) {
			continue // copied by an earlier run
		} else if held != nil {
			return nil, fmt.Errorf("the ledger holds %s with %d versions, the database %d: it is not a copy; nothing more was copied", k, held.Versions, len(p.versions))
		}
		if err := m.led.Import(ctx, p.e, p.versions); err != nil {
			return nil, fmt.Errorf("%s could not be copied (%v); %d records copied, the rest not, and nothing is switched", k, err, copied)
		}
		// The events of these versions were said when they were written in the database: none is owed.
		_, _ = m.st.State(sentState).Put(sentKey(k), len(p.versions))
		copied++
		if say != nil && copied%100 == 0 {
			say(fmt.Sprintf("copied %d of %d records", copied, len(plans)))
		}
	}
	if f, err := m.db.Findings(ctx); err == nil {
		_ = m.led.SetFindings(ctx, f)
	}
	if told, err := m.db.ToldOf(ctx, ""); err == nil {
		for _, t := range told {
			_, _ = m.led.Told(ctx, t)
		}
	}
	problems, err := m.Check(ctx)
	if err != nil {
		return nil, err
	}
	out.Problems = problems
	if len(problems) > 0 {
		out.Says = fmt.Sprintf("The copy of %d records is not clean (%d difference(s)); nothing is switched and the database stays the source.", out.Records, len(problems))
		return out, nil
	}
	if _, err := m.kv().Put(switchKey, SourceLedger); err != nil {
		return nil, fmt.Errorf("the check is clean and the switch could not be kept (%v): the database stays the source", err)
	}
	m.mu.Lock()
	m.active = SourceLedger
	m.mu.Unlock()
	out.Switched, out.Source = true, SourceLedger
	out.Says = fmt.Sprintf("%d records (%d versions) are copied and the check is clean: they are served from the ledger now, and the database is written beside it until it is dropped on the operator's yes.", out.Records, out.Versions)
	return out, nil
}

// MostDifferences is how many differences a check names.
const MostDifferences = 50

// Check compares the database with the ledger, record by record: the versions, the served and the proposed version whole,
// every section's text. It answers every difference, at most MostDifferences, and keeps the result in the state; a clean
// check is an empty answer.
func (m *movableStore) Check(ctx context.Context) ([]string, error) {
	dbAll, err := m.db.All(ctx)
	if err != nil {
		return nil, fmt.Errorf("the database cannot be read: %v", err)
	}
	var diffs []string
	add := func(format string, a ...any) {
		if len(diffs) < MostDifferences {
			diffs = append(diffs, fmt.Sprintf(format, a...))
		}
	}
	same := func(a, b *Record) bool {
		if a == nil || b == nil {
			return a == b
		}
		x, _ := json.Marshal(a)
		y, _ := json.Marshal(b)
		return string(x) == string(y)
	}
	seen := map[Key]bool{}
	for _, d := range dbAll {
		k := d.Current.Key()
		seen[k] = true
		l, err := m.led.Get(ctx, k)
		if err != nil {
			return nil, err
		}
		switch {
		case l == nil:
			add("%s is in the database and not in the ledger", k)
			continue
		case l.Versions != d.Versions || l.CurrentVersion != d.CurrentVersion || l.PendingVersion != d.PendingVersion:
			add("%s: the ledger has %d versions (served %d, proposed %d), the database %d (served %d, proposed %d)", k, l.Versions, l.CurrentVersion, l.PendingVersion, d.Versions, d.CurrentVersion, d.PendingVersion)
		case !same(l.Current, d.Current):
			add("%s: the served version differs", k)
		case !same(l.Pending, d.Pending):
			add("%s: the proposed version differs", k)
		}
		dv, err := m.db.Versions(ctx, k)
		if err != nil {
			return nil, err
		}
		lv, err := m.led.Versions(ctx, k)
		if err != nil {
			return nil, err
		}
		if len(dv) == len(lv) {
			for i := range dv {
				if dv[i].N != lv[i].N || dv[i].Why != lv[i].Why || !same(dv[i].Record, lv[i].Record) {
					add("%s: version %d differs", k, dv[i].N)
					break
				}
			}
		}
	}
	if all, err := m.led.All(ctx); err == nil {
		for _, l := range all {
			if k := l.Current.Key(); !seen[k] {
				add("%s is in the ledger and not in the database", k)
			}
		}
	}
	if e, err := m.kv().Get(behindKey); err == nil && e != nil {
		add("the database is behind: %s", strings.TrimSpace(string(e.Value)))
	}
	sort.Strings(diffs)
	_, _ = m.kv().Put(checkKey, map[string]any{"differences": diffs, "at": time.Now().UTC(), "records": len(dbAll)})
	return diffs, nil
}

// Import writes a record's versions into the ledger as `import` operations, the old author kept apart as imported and the
// ledger's author the module's own account (ADR 0597 §7), each section's text once; the last operation carries the head.
func (s *ledgerStore) Import(ctx context.Context, e *Entry, versions []Version) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := e.Current.Key()
	if len(versions) == 0 {
		return fmt.Errorf("%s has no version to copy", k)
	}
	var known []sectionRef
	var head *entryHead
	for i, v := range versions {
		refs, err := s.write(ctx, k, v.Record, known)
		if err != nil {
			return err
		}
		known = append(known, refs...)
		h := &entryHead{Versions: v.N}
		// Each operation's head is the one the database held at that version; the last is the entry as it stands.
		if i == len(versions)-1 {
			h.CurrentVersion, h.PendingVersion = e.CurrentVersion, e.PendingVersion
		} else {
			h.CurrentVersion = v.N
		}
		for _, side := range []struct {
			n  int
			to **storedRecord
		}{{h.CurrentVersion, &h.Current}, {h.PendingVersion, &h.Pending}} {
			if side.n == 0 {
				continue
			}
			for _, w := range versions {
				if w.N == side.n {
					r, err := s.write(ctx, k, w.Record, known)
					if err != nil {
						return err
					}
					*side.to = ptr(stripped(w.Record, r))
				}
			}
		}
		head = h
		body := map[string]any{"kind": k.Kind, "number": k.Number, "entry": head,
			"version":          versionHead{N: v.N, By: v.By, ImportedBy: v.By, At: v.At, Why: v.Why, Insight: v.Insight, Record: stripped(v.Record, refs)},
			"pending-sections": map[string]any{}}
		seq, err := s.apply(ctx, k, OpImport, body)
		if err != nil {
			return err
		}
		s.heads[k] = &headState{Kind: k.Kind, Number: k.Number, Entry: head, Sequence: seq}
	}
	s.hold(k)
	s.indexed(e.Current, true)
	return nil
}