package main

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

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/catalogue"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// A module declares the data it holds, and the mesh protects and watches it from that declaration
// (novox/hq ADR 0233).
//
// The self-check's D13 composes what every machine declares, asks each machine's backup holder what
// it measured of every item — size, newest write, newest good backup, the redundant storage it is on —
// and keeps both. From that, and from what every provider says it holds for its consumers, it raises,
// each URGENT for what is irreplaceable and a WARNING for what is valuable (the operator's ranking):
//
//   - `data-shrank`: an item holds less than half of its largest size in seven days, and at least
//     shrinkFloor less; `data-missing`: its path is gone;
//   - `empty-replacement`: an item, or a consumer's data at a provider, is less than half the size of a
//     copy of the same thing kept elsewhere — on 2026-10-05 five applications ran for twenty hours on
//     empty databases while their real ones sat on another machine (issue 273);
//   - `data-held-twice` (warning): a consumer has active data at two providers and their sizes cannot
//     be compared;
//   - `data-quiet`: an item said to be written all the time has not been, within its bound;
//   - `backup-stale`: an item's newest good backup is older than its bound, or there is none;
//   - `array-degraded`: the redundant storage an item is on is not healthy, or cannot be read;
//     `protection-missing`: an item said to be protected by redundancy is on storage that is not;
//   - `cleanup-waiting` (warning): an item retired more than thirty days, waiting for a person.
//
// And it retires: an irreplaceable or valuable item in a module's own directory that its machine no
// longer declares — its module unassigned — is kept, marked retired with when and why, and listed by
// `cleanup list` until `cleanup delete` removes it. The node-engine never deletes a directory with
// anything in it; this is the record of what it kept. An operator's path is never retired or deleted.

// The condition kinds of D13.
const (
	kindDataShrank       = "data-shrank"
	kindEmptyReplacement = "empty-replacement"
	kindDataHeldTwice    = "data-held-twice"
	kindDataQuiet        = "data-quiet"
	kindBackupStale      = "backup-stale"
	kindDataUnmeasured   = "data-unmeasured"
	// kindDataMissing is a watched item whose path is gone.
	kindDataMissing = "data-missing"
	// kindArrayDegraded is redundant storage watched data is on that is not healthy, or cannot be read.
	kindArrayDegraded = "array-degraded"
	// kindProtectionMissing is an item said to be protected by redundancy, on storage that is not.
	kindProtectionMissing = "protection-missing"
)

// probeDataID is the self-check's id for this probe.
const probeDataID = "D13"

// The bounds the findings are read against.
var (
	// shrinkWindow is how far back the largest size is looked for.
	shrinkWindow = 7 * 24 * time.Hour
	// shrinkFloor is the least loss that is worth saying: two empty databases differ by a few
	// megabytes, and half of almost nothing is noise.
	shrinkFloor int64 = 16 << 20
	// dataAsk is how long one machine's holder, or one provider, is given to answer.
	dataAsk = 8 * time.Second
)

// keyOfItem is one item's condition id: its machine, module and item.
func keyOfItem(machine, module, item string) string { return machine + "." + module + "." + item }

// holderAnswer is what a node-backup holder's `backed-up` says of one module (ADR 0233 adds Data).
type holderAnswer struct {
	Module string       `json:"module"`
	Data   []holderItem `json:"data"`
}

// holderItem is one item as the holder measured it.
type holderItem struct {
	Item       string     `json:"item"`
	Class      string     `json:"class"`
	Path       string     `json:"path"`
	SizeBytes  *int64     `json:"size_bytes"`
	LastWrite  *time.Time `json:"last_write"`
	MeasuredAt *time.Time `json:"measured_at"`
	LastBackup *time.Time `json:"last_backup"`
	Error      string     `json:"error,omitempty"`
	// Precision is what the size is: exact, a dataset's, partial, or none (ADR 0233).
	Precision string `json:"precision,omitempty"`
	// Redundancy is the redundant storage the item is on, where the holder could tell (ADR 0233).
	Redundancy *inventory.Redundancy `json:"redundancy,omitempty"`
}

// readHolder reads a holder's answer into measurements by module and item.
func readHolder(raw json.RawMessage) (map[string]map[string]inventory.Measurement, error) {
	var modules []holderAnswer
	if err := json.Unmarshal(raw, &modules); err != nil {
		return nil, fmt.Errorf("its answer is not readable: %w", err)
	}
	out := map[string]map[string]inventory.Measurement{}
	for _, m := range modules {
		for _, it := range m.Data {
			if out[m.Module] == nil {
				out[m.Module] = map[string]inventory.Measurement{}
			}
			out[m.Module][it.Item] = inventory.Measurement{Path: it.Path, Size: it.SizeBytes, LastWrite: it.LastWrite,
				MeasuredAt: it.MeasuredAt, LastBackup: it.LastBackup, Error: it.Error, Redundancy: it.Redundancy,
				Precision: it.Precision}
		}
	}
	return out, nil
}

// declaredOn is every data item a machine's composition declares, and the module there that holds
// node-backup to measure them — empty for none.
func declaredOn(plan catalogue.Resolution) ([]inventory.DeclaredData, string) {
	var out []inventory.DeclaredData
	held := ""
	for _, m := range plan.Modules {
		for _, c := range m.Claims {
			if s, known := catalogue.SeatNamed(c.Name); known && s.Name == catalogue.BackupSeat {
				held = m.Module
			}
		}
		for _, it := range m.DataItems() {
			out = append(out, inventory.DeclaredData{Module: m.Module, Item: it.ID, Class: it.Class,
				Owned: it.OwnedByModule(), Protection: it.Protection()})
		}
	}
	return out, held
}

// consumerCopy is one provider's account of one consumer: where, how big, and whether still active.
type consumerCopy struct {
	Node, Module, Consumer string
	Size                   *int64
	Retired                bool
	// Class is how precious the consumer's data is: the stricter of what the provider keeps for its
	// consumers and what the consumer says it keeps there (`kept-by`).
	Class string
}

// probeData is D13.
func probeData(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	if d.js == nil {
		return nil, errors.New("no bus to ask the machines over")
	}
	open := d.open
	shelf, err := open.inventory.Catalogue(ctx)
	if err != nil {
		return nil, err
	}
	nodes, err := open.inventory.Nodes(ctx)
	if err != nil {
		return nil, err
	}
	heard := heardMachines(d)
	now := time.Now()
	delivered, err := readDeliveries(ctx, open.inventory)
	if err != nil {
		return nil, err
	}

	type machine struct {
		name     string
		declared []inventory.DeclaredData
		held     bool
		measured map[string]map[string]inventory.Measurement
		askErr   error
	}
	var machines []*machine
	for _, n := range nodes {
		plan, _, err := planFor(ctx, open, n.Name)
		if err != nil {
			if ctx.Err() != nil {
				return nil, ctx.Err()
			}
			// A machine that cannot be worked out declares nothing this run — which is not the same as
			// declaring nothing: retiring its data on that would be acting on an unreadable result.
			continue
		}
		declared, holder := declaredOn(plan)
		// **A holder is asked only once its machine has been sent it and had time to report** (novox/hq
		// issue 275): assigned and not pushed yet, it is not there to answer, and "did not say what it
		// measured" about it was a warning for a push nobody had made yet.
		held := holder != "" && delivered[n.Name].settled(holder, now)
		machines = append(machines, &machine{name: n.Name, declared: declared, held: held})
	}
	// Every holder asked at once, as D8 asks every ban list.
	var wg sync.WaitGroup
	for _, m := range machines {
		if !m.held || !heard[m.name] {
			continue
		}
		wg.Add(1)
		go func(m *machine) {
			defer wg.Done()
			asking, cancel := context.WithTimeout(ctx, dataAsk)
			defer cancel()
			raw, err := askSeatTool(asking, d.js.Conn(), catalogue.BackupSeat, "backed-up", m.name)
			if err == nil {
				m.measured, err = readHolder(raw)
			}
			m.askErr = err
		}(m)
	}
	wg.Wait()

	var out []conditions.Observation
	for _, m := range machines {
		if m.askErr != nil {
			out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: m.name, Token: kindDataUnmeasured,
				Kind: kindDataUnmeasured, Machine: m.name, Severity: conditions.Warning, Confirm: true,
				Summary: fmt.Sprintf("%s's backup holder did not say what it measured of the data declared there, so "+
					"nothing about that data is known", m.name),
				Said: firstLine(m.askErr.Error())})
		}
		why := fmt.Sprintf("no longer declared on %s: its module was unassigned there, or is no longer pulled in", m.name)
		change, err := open.inventory.RecordData(ctx, m.name, m.declared, m.measured, why, now)
		if err != nil {
			return nil, fmt.Errorf("what %s holds could not be kept: %w", m.name, err)
		}
		for _, r := range change.Retired {
			log.Printf("data: %s of %s on %s RETIRED, kept at %s: %s — `cleanup list` shows it, and only `cleanup "+
				"delete` removes it (novox/hq ADR 0233)", r.Item, r.Module, r.Machine, orUnknownPath(r.Path), why)
		}
		for _, r := range change.Reenabled {
			log.Printf("data: %s of %s on %s is declared again, no longer retired", r.Item, r.Module, r.Machine)
		}
	}

	records, err := open.inventory.Data(ctx)
	if err != nil {
		return nil, err
	}
	peaks, err := open.inventory.DataPeaks(ctx, now.Add(-shrinkWindow))
	if err != nil {
		return nil, err
	}
	bindings, err := open.inventory.Bindings(ctx)
	if err != nil {
		return nil, err
	}
	upgraded, keptBy := keptByClasses(bindings, shelf)
	copies, err := consumerCopies(ctx, d.js.Conn(), open.inventory, shelf, keptBy)
	if err != nil {
		return nil, err
	}
	out = append(out, dataFindings(records, peaks, shelf, copies, upgraded, now)...)
	return out, nil
}

func orUnknownPath(p string) string {
	if p == "" {
		return "a path its backup holder never named"
	}
	return p
}

// consumerCopies asks every provider of a provision whose consumers' data is kept what it holds, at
// once. One that cannot answer is passed over: it says nothing about any copy, which is not a finding.
func consumerCopies(ctx context.Context, conn *nats.Conn, inv *inventory.Inventory,
	shelf map[string]catalogue.Manifest, keptBy map[string]string) ([]consumerCopy, error) {
	instances, err := providerInstances(ctx, inv)
	if err != nil {
		return nil, err
	}
	var asked []providerInstance
	for _, p := range instances {
		m := shelf[p.Module]
		keeps := false
		for provision := range m.Grants {
			keeps = keeps || m.KeepsConsumerData(provision)
		}
		if keeps {
			asked = append(asked, p)
		}
	}
	states := make([]*link.RetirementState, len(asked))
	var wg sync.WaitGroup
	for i, p := range asked {
		wg.Add(1)
		go func(i int, p providerInstance) {
			defer wg.Done()
			asking, cancel := context.WithTimeout(ctx, dataAsk)
			defer cancel()
			if s, err := askRetirement(asking, conn, p); err == nil {
				states[i] = &s
			}
		}(i, p)
	}
	wg.Wait()
	var out []consumerCopy
	for i, p := range asked {
		s := states[i]
		if s == nil {
			continue
		}
		class := consumersClass(shelf[p.Module])
		for _, c := range s.Held {
			cp := consumerCopy{Node: p.Node, Module: p.Module, Consumer: c,
				Class: catalogue.StricterClass(class, keptBy[p.Module+"/"+c])}
			if size, ok := s.HeldSizes[c]; ok && size >= 0 {
				size := size
				cp.Size = &size
			}
			out = append(out, cp)
		}
		for _, r := range s.Retired {
			if r.Kind != "" && r.Kind != "consumer" {
				continue
			}
			cp := consumerCopy{Node: p.Node, Module: p.Module, Consumer: r.Consumer, Retired: true,
				Class: catalogue.StricterClass(class, keptBy[p.Module+"/"+r.Consumer])}
			if r.SizeBytes != nil && *r.SizeBytes >= 0 {
				cp.Size = r.SizeBytes
			}
			out = append(out, cp)
		}
	}
	return out, nil
}

// severityOf is how loud a finding about data of a class is: urgent for what is irreplaceable, a warning
// for anything else watched (the operator's ranking, ADR 0233).
func severityOf(class string) conditions.Severity {
	if class == catalogue.ClassIrreplaceable {
		return conditions.Urgent
	}
	return conditions.Warning
}

// consumersClass is the most precious class a provider keeps any of its consumers' data as.
func consumersClass(m catalogue.Manifest) string {
	class := catalogue.ClassNone
	for provision := range m.Grants {
		if c, ok := m.ConsumerDataOf(provision); ok {
			class = catalogue.StricterClass(class, c.Class)
		} else if m.KeepsConsumerData(provision) {
			class = catalogue.StricterClass(class, catalogue.ClassValuable)
		}
	}
	return class
}

// keptByClasses is what consumers say of the data they keep with their providers (`kept-by`), read
// through where each is bound: by provider module and consumer identity, the class of that consumer's
// data there; and by provider item key (machine/module/item), the class the item holding it is held to.
func keptByClasses(bindings []inventory.Binding, shelf map[string]catalogue.Manifest) (map[string]string, map[string]string) {
	upgraded, keptBy := map[string]string{}, map[string]string{}
	for _, b := range bindings {
		m, ok := shelf[b.Consumer]
		if !ok {
			continue
		}
		k, said := m.KeptByOf(b.Provision)
		if !said {
			continue
		}
		identity := catalogue.ConsumerIdentity(b.Machine, catalogue.IdentitySource(m.Slug, m.Module))
		key := b.Provider.Module + "/" + identity
		keptBy[key] = catalogue.StricterClass(keptBy[key], k.Class)
		if pc, ok := shelf[b.Provider.Module].ConsumerDataOf(b.Provision); ok && pc.In != "" {
			if _, own := shelf[b.Provider.Module].DataItem(pc.In); own {
				item := b.Provider.Node + "/" + b.Provider.Module + "/" + pc.In
				upgraded[item] = catalogue.StricterClass(upgraded[item], k.Class)
			}
		}
	}
	return upgraded, keptBy
}

// dataFindings is every condition the data on record raises now. A function of what is known, so the
// incident's shape is tested without a mesh. upgraded is the class an item is held to where a consumer
// of its module keeps data in it more precious than its own class says (`kept-by`), by its key.
func dataFindings(records []inventory.DataRecord, peaks map[string]int64, shelf map[string]catalogue.Manifest,
	copies []consumerCopy, upgraded map[string]string, now time.Time) []conditions.Observation {
	var out []conditions.Observation
	byItem := map[string][]inventory.DataRecord{}
	arrays := map[string][]inventory.DataRecord{}
	type shrunk struct {
		machine, dataset, class string
		size, peak              int64
		items                   []string
	}
	shrunkDatasets := map[string]shrunk{}
	for _, r := range records {
		if r.DeletedAt != nil {
			continue
		}
		class := catalogue.StricterClass(r.Class, upgraded[r.Key()])
		r.Class = class
		byItem[r.Module+"/"+r.Item] = append(byItem[r.Module+"/"+r.Item], r)
		item, declared := shelf[r.Module].DataItem(r.Item)
		id := keyOfItem(r.Machine, r.Module, r.Item)
		if r.Retired() {
			if now.Sub(*r.RetiredAt) > cleanupAfter {
				out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: "cleanup",
					Kind: kindCleanupWaiting, Machine: r.Machine, Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
					Summary: fmt.Sprintf("%s of %s on %s (%s, %s) has been retired %d days — kept where it was since %s; "+
						"`cleanup delete %s %s %s --why …` once a person has decided, or assign %s there again",
						r.Item, r.Module, r.Machine, r.Class, sizeWords(r.Size), int(now.Sub(*r.RetiredAt).Hours()/24),
						r.RetiredWhy, r.Machine, r.Module, r.Item, r.Module),
					Said: "kept at " + orUnknownPath(r.Path)})
			}
			continue
		}
		if !catalogue.Watched(class) {
			continue
		}
		severity := severityOf(class)
		if r.Redundancy != nil {
			where := r.Machine + "/" + r.Redundancy.Kind + ":" + r.Redundancy.Where
			arrays[where] = append(arrays[where], r)
		} else if declared && item.Redundancy != "" && r.MeasuredAt != nil && r.MeasureError == "" {
			out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindProtectionMissing,
				Kind: kindProtectionMissing, Machine: r.Machine, Severity: severity, Resolver: conditions.ResolverOperator,
				Summary: fmt.Sprintf("%s of %s on %s (%s) is said to be protected by the redundancy of the storage it is on, "+
					"and it is on nothing the backup holder can read as redundant: it has no protection the mesh can see",
					r.Item, r.Module, r.Machine, class),
				Said: orUnknownPath(r.Path) + " is on no redundant storage the backup holder can read"})
		}
		if r.MeasureError != "" && strings.Contains(r.MeasureError, "does not exist") {
			out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindDataMissing,
				Kind: kindDataMissing, Machine: r.Machine, Severity: severity, Resolver: conditions.ResolverOperator,
				Summary: fmt.Sprintf("%s of %s on %s (%s) is gone: where it was declared, nothing exists any more", r.Item,
					r.Module, r.Machine, class),
				Said: orUnknownPath(r.Path) + " does not exist: " + firstLine(r.MeasureError)})
		} else if peak, ok := peaks[r.Key()]; ok && r.Size != nil && inventory.Comparable(r.Precision) &&
			*r.Size*2 < peak && peak-*r.Size >= shrinkFloor && inventory.Dataset(r.Precision) != "" {
			// Several items on one dataset share its size: one condition for the dataset, as loud as the
			// most precious item on it.
			k := r.Machine + "/" + inventory.Dataset(r.Precision)
			ds := shrunkDatasets[k]
			ds.machine, ds.dataset, ds.size, ds.peak = r.Machine, inventory.Dataset(r.Precision), *r.Size, peak
			ds.class = catalogue.StricterClass(ds.class, class)
			ds.items = append(ds.items, r.Module+"/"+r.Item)
			shrunkDatasets[k] = ds
		} else if peak, ok := peaks[r.Key()]; ok && r.Size != nil && inventory.Comparable(r.Precision) &&
			*r.Size*2 < peak && peak-*r.Size >= shrinkFloor {
			out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindDataShrank,
				Kind: kindDataShrank, Machine: r.Machine, Severity: severity, Resolver: conditions.ResolverOperator,
				Summary: fmt.Sprintf("%s of %s on %s (%s) shrank to %s from %s within %d days — more than half of what it "+
					"held is gone. If that was meant, silence this with why; if not, `node-backup.restore` puts the last "+
					"good copy beside it", r.Item, r.Module, r.Machine, class, sizeWords(r.Size), sizeWords(&peak),
					int(shrinkWindow.Hours()/24))})
		}
		if !declared {
			continue
		}
		if within := item.ActiveWithin(); within > 0 && r.LastWrite != nil && now.Sub(*r.LastWrite) > within {
			out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindDataQuiet,
				Kind: kindDataQuiet, Machine: r.Machine, Severity: severity,
				Summary: fmt.Sprintf("%s of %s on %s is written all the time, and has not been since %s (its bound is %s): "+
					"whatever writes it has stopped", r.Item, r.Module, r.Machine, r.LastWrite.UTC().Format(time.RFC3339),
					within)})
		}
		// A backup is required of what is irreplaceable and copied; of what is valuable it is the standard
		// plan, said only where the machine was measured — where a holder is there to take it.
		if item.BackedUp() && (class == catalogue.ClassIrreplaceable || r.MeasuredAt != nil) {
			within := item.BackupWithin()
			switch {
			case r.LastBackup == nil && now.Sub(r.FirstSeen) > within:
				out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindBackupStale,
					Kind: kindBackupStale, Machine: r.Machine, Severity: severity,
					Summary: fmt.Sprintf("%s of %s on %s is %s and has no good backup on record, %s after it was first "+
						"declared — is node-backup held there, and do its nights succeed? (`node-backup.backed-up`)",
						r.Item, r.Module, r.Machine, class, now.Sub(r.FirstSeen).Round(time.Hour))})
			case r.LastBackup != nil && now.Sub(*r.LastBackup) > within:
				out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: id, Token: kindBackupStale,
					Kind: kindBackupStale, Machine: r.Machine, Severity: severity,
					Summary: fmt.Sprintf("%s of %s on %s is %s and its newest good backup is from %s, older than its bound "+
						"of %s", r.Item, r.Module, r.Machine, class, r.LastBackup.UTC().Format(time.RFC3339), within)})
			}
		}
	}
	for _, k := range keysSorted(shrunkDatasets) {
		ds := shrunkDatasets[k]
		out = append(out, conditions.Observation{Scope: conditions.ScopeMachine,
			ID: ds.machine + ".dataset." + strings.ReplaceAll(ds.dataset, "/", "-"), Token: kindDataShrank,
			Kind: kindDataShrank, Machine: ds.machine, Severity: severityOf(ds.class), Resolver: conditions.ResolverOperator,
			Summary: fmt.Sprintf("the dataset on %s that holds %s shrank to %s from %s within %d days — more than half of "+
				"what it held is gone", ds.machine, strings.Join(ds.items, ", "), sizeWords(&ds.size), sizeWords(&ds.peak),
				int(shrinkWindow.Hours()/24)),
			Said: "the dataset " + ds.dataset})
	}
	// The redundant storage watched data is on: one condition per array, as loud as the most precious
	// item on it — the array, not each item, is what degrades.
	for _, where := range keysSorted(arrays) {
		rs := arrays[where]
		red := rs[0].Redundancy
		if red.Healthy != nil && *red.Healthy {
			continue
		}
		class, machine := catalogue.ClassValuable, rs[0].Machine
		var names []string
		for _, r := range rs {
			class = catalogue.StricterClass(class, r.Class)
			names = append(names, r.Module+"/"+r.Item)
		}
		state := "could not be read"
		if red.Healthy != nil {
			state = "is NOT healthy"
		}
		out = append(out, conditions.Observation{Scope: conditions.ScopeMachine,
			ID:    machine + ".array." + strings.NewReplacer("/", "-", ":", "-").Replace(red.Kind+"-"+red.Where),
			Token: kindArrayDegraded, Kind: kindArrayDegraded, Machine: machine, Severity: severityOf(class),
			Resolver: conditions.ResolverOperator,
			Summary: fmt.Sprintf("the %s storage on %s that protects %s %s", red.Kind, machine, strings.Join(names, ", "),
				state),
			Said: fmt.Sprintf("the %s storage %s %s: %s", red.Kind, red.Where, state, firstLine(red.Said))})
	}
	// The same item on several machines: a copy that is in use and far smaller than one kept elsewhere is
	// an empty replacement. Only against a retired copy — a module running on two machines on purpose
	// keeps two different sets of data.
	for _, key := range keysSorted(byItem) {
		rs := byItem[key]
		for _, a := range rs {
			if a.Retired() || a.Size == nil || !catalogue.Watched(a.Class) || !inventory.Comparable(a.Precision) {
				continue
			}
			for _, o := range rs {
				if o.Machine == a.Machine || !o.Retired() || o.Size == nil || !inventory.Comparable(o.Precision) ||
					!replacedByLess(*a.Size, *o.Size) {
					continue
				}
				out = append(out, conditions.Observation{Scope: conditions.ScopeMachine,
					ID: keyOfItem(a.Machine, a.Module, a.Item), Token: kindEmptyReplacement, Kind: kindEmptyReplacement,
					Machine: a.Machine, Also: []string{o.Machine}, Severity: severityOf(a.Class), Resolver: conditions.ResolverOperator,
					Summary: fmt.Sprintf("%s of %s on %s holds %s, and the copy %s kept on %s holds %s: %s is running on "+
						"an empty replacement of its data. Move the data, or assign it back where its data is",
						a.Item, a.Module, a.Machine, sizeWords(a.Size), o.Module, o.Machine, sizeWords(o.Size), a.Module)})
				break
			}
		}
	}
	out = append(out, consumerFindings(copies)...)
	return out
}

// replacedByLess is whether a copy in use is an empty replacement of a copy kept elsewhere: less than
// half of it, and at least shrinkFloor less.
func replacedByLess(inUse, kept int64) bool {
	return inUse*2 < kept && kept-inUse >= shrinkFloor
}

// consumerFindings is the same question of consumers' data at providers: one consumer, the same
// provider module on two machines.
func consumerFindings(copies []consumerCopy) []conditions.Observation {
	by := map[string][]consumerCopy{}
	for _, c := range copies {
		k := c.Module + "/" + c.Consumer
		by[k] = append(by[k], c)
	}
	var out []conditions.Observation
	for _, k := range keysSorted(by) {
		cs := by[k]
		if len(cs) < 2 {
			continue
		}
		found := false
		for _, a := range cs {
			if a.Retired || a.Size == nil {
				continue
			}
			for _, o := range cs {
				if o.Node == a.Node || o.Size == nil || !replacedByLess(*a.Size, *o.Size) {
					continue
				}
				state := "active"
				if o.Retired {
					state = "retired"
				}
				out = append(out, conditions.Observation{Scope: conditions.ScopeProvider,
					ID: a.Module + "." + a.Node + "." + a.Consumer, Token: kindEmptyReplacement, Kind: kindEmptyReplacement,
					Machine: a.Node, Also: []string{o.Node}, Severity: severityOf(catalogue.StricterClass(a.Class, o.Class)),
					Resolver: conditions.ResolverOperator,
					Summary: fmt.Sprintf("%s's data at %s on %s holds %s, and its %s copy at %s on %s holds %s: the "+
						"consumer is using an empty replacement of its data (issue 273's shape). Pin it back to %s, or move "+
						"the data first", a.Consumer, a.Module, a.Node, sizeWords(a.Size), state, o.Module, o.Node,
						sizeWords(o.Size), o.Node)})
				found = true
				break
			}
			if found {
				break
			}
		}
		if found {
			continue
		}
		var active []consumerCopy
		for _, c := range cs {
			if !c.Retired {
				active = append(active, c)
			}
		}
		if len(active) >= 2 {
			var where []string
			var also []string
			for _, c := range active {
				where = append(where, c.Node+" ("+sizeWords(c.Size)+")")
				also = append(also, c.Node)
			}
			out = append(out, conditions.Observation{Scope: conditions.ScopeProvider,
				ID: active[0].Module + "." + active[0].Consumer, Token: kindDataHeldTwice, Kind: kindDataHeldTwice,
				Machine: active[0].Node, Also: also[1:], Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
				Summary: fmt.Sprintf("%s has active data at %s on %d machines — %s — and only one is the one it uses",
					active[0].Consumer, active[0].Module, len(active), strings.Join(where, ", "))})
		}
	}
	return out
}

func keysSorted[V any](m map[string]V) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// ---- the `data` verb ---------------------------------------------------------------------------

const dataUsage = "data [--json] [--machine <name>] [--retired]"

// dataRow is one item as `data` lists it.
type dataRow struct {
	Machine    string `json:"machine"`
	Module     string `json:"module"`
	Item       string `json:"item"`
	Class      string `json:"class"`
	Path       string `json:"path,omitempty"`
	Protection string `json:"protection,omitempty"`
	// Array is the redundant storage it is on and its state, where its holder could tell.
	Array      string `json:"array,omitempty"`
	Unmeasured string `json:"unmeasured,omitempty"`
	// Precision says what the size is: exact, a dataset's whole size, partial, or none.
	Precision  string `json:"precision,omitempty"`
	SizeBytes  *int64 `json:"size-bytes,omitempty"`
	LastWrite  string `json:"last-write,omitempty"`
	MeasuredAt string `json:"measured-at,omitempty"`
	LastBackup string `json:"last-backup,omitempty"`
	BackupDue  string `json:"backup-within,omitempty"`
	Retired    string `json:"retired,omitempty"`
	RetiredWhy string `json:"retired-why,omitempty"`
	Deleted    string `json:"deleted,omitempty"`
}

// dataCommand is `data`: every item every machine declares, or held retired, as the self-check last
// found it.
func dataCommand(ctx context.Context, args []string) error {
	set := flag.NewFlagSet("data", flag.ContinueOnError)
	asJSON := set.Bool("json", false, "as data")
	only := set.String("machine", "", "one machine")
	retiredOnly := set.Bool("retired", false, "only what is retired")
	if rest, err := parseAround(set, args); err != nil {
		return err
	} else if len(rest) > 0 {
		return errors.New(dataUsage)
	}
	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	records, err := open.inventory.Data(ctx)
	if err != nil {
		return err
	}
	shelf, err := open.inventory.Catalogue(ctx)
	if err != nil {
		return err
	}
	rows := dataRows(records, shelf, *only, *retiredOnly)
	if *asJSON {
		return printJSON(map[string]any{"data": rows})
	}
	if len(rows) == 0 {
		fmt.Println("no data on record: the self-check (D13) records what each machine declares on its next run")
		return nil
	}
	for _, r := range rows {
		state := ""
		switch {
		case r.Deleted != "":
			state = "  DELETED " + r.Deleted
		case r.Retired != "":
			state = "  RETIRED " + r.Retired + " — " + r.RetiredWhy
		}
		fmt.Printf("%s %s/%s  %s  %s  %s  protected by %s%s\n", r.Machine, r.Module, r.Item, r.Class,
			sizeWords(r.SizeBytes), orUnknownPath(r.Path), orNothingWord(r.Protection), state)
		if r.Array != "" {
			fmt.Printf("    on %s\n", r.Array)
		}
		if r.Precision != "" && r.Precision != "exact" {
			fmt.Printf("    size: %s\n", r.Precision)
		}
		if r.Unmeasured != "" {
			fmt.Printf("    not measured: %s\n", r.Unmeasured)
		}
		if r.Class == catalogue.ClassCache {
			continue
		}
		fmt.Printf("    last write %s, measured %s, last backup %s%s\n", orNever(r.LastWrite), orNever(r.MeasuredAt),
			orNever(r.LastBackup), within(r.BackupDue))
	}
	return nil
}

func dataRows(records []inventory.DataRecord, shelf map[string]catalogue.Manifest, only string, retiredOnly bool) []dataRow {
	rows := []dataRow{}
	stamp := func(t *time.Time) string {
		if t == nil {
			return ""
		}
		return t.UTC().Format(time.RFC3339)
	}
	for _, r := range records {
		if only != "" && r.Machine != only {
			continue
		}
		if retiredOnly && !r.Retired() {
			continue
		}
		row := dataRow{Machine: r.Machine, Module: r.Module, Item: r.Item, Class: r.Class, Path: r.Path,
			Protection: r.Protection, Unmeasured: r.MeasureError, Precision: r.Precision, SizeBytes: r.Size, LastWrite: stamp(r.LastWrite), MeasuredAt: stamp(r.MeasuredAt),
			LastBackup: stamp(r.LastBackup), Retired: stamp(r.RetiredAt), RetiredWhy: r.RetiredWhy,
			Deleted: stamp(r.DeletedAt)}
		if it, ok := shelf[r.Module].DataItem(r.Item); ok && it.BackedUp() {
			row.BackupDue = it.BackupWithin().String()
		}
		if red := r.Redundancy; red != nil {
			state := "state unread"
			if red.Healthy != nil && *red.Healthy {
				state = "healthy"
			} else if red.Healthy != nil {
				state = "NOT HEALTHY"
			}
			row.Array = red.Kind + " " + red.Where + ", " + state
		}
		rows = append(rows, row)
	}
	return rows
}

func orNothingWord(s string) string {
	if s == "" || s == "none" {
		return "nothing"
	}
	return s
}

func orNever(s string) string {
	if s == "" {
		return "never"
	}
	return s
}

func within(s string) string {
	if s == "" {
		return " (not backed up)"
	}
	return " (bound " + s + ")"
}

// ---- cleanup of retired own data ---------------------------------------------------------------

// The tools a node-backup holder serves to delete one retired item (novox/hq ADR 0233): the first
// takes a last restore point of it, tagged as retired, and only then removes it — in the background,
// because a large item outlasts any call — and the second says how that went. Module tools, not seat
// verbs: only the controller's `cleanup delete` calls them, as it calls a provider's provisioner_delete.
const (
	ToolDeleteRetired  = "backup_delete_retired"
	ToolDeletedOutcome = "backup_deleted"
)

// deletionWait is how long `cleanup delete` follows a deletion before handing it back to the person.
var deletionWait = 8 * time.Minute

// deletionPoll is how often it asks.
var deletionPoll = 5 * time.Second

// deletion is a holder's account of one deletion.
type deletion struct {
	Started  bool   `json:"started"`
	Running  bool   `json:"running"`
	Done     bool   `json:"done"`
	OK       bool   `json:"ok"`
	Snapshot string `json:"snapshot"`
	Error    string `json:"error"`
}

// retiredData is every retired item on record, as `cleanup list` shows them.
func retiredData(records []inventory.DataRecord, now time.Time) []retiredRow {
	var out []retiredRow
	for _, r := range records {
		if !r.Retired() {
			continue
		}
		out = append(out, retiredRow{Node: r.Machine, Module: r.Module, Consumer: r.Item, Kind: retiredDataKind,
			RetiredAt: r.RetiredAt.UTC().Format(time.RFC3339), AgeDays: int(now.Sub(*r.RetiredAt).Hours() / 24),
			SizeBytes: r.Size, Why: r.RetiredWhy, Path: r.Path, Class: r.Class})
	}
	return out
}

// retiredDataKind is what `cleanup list` calls a module's own retired data, beside a provider's consumer.
const retiredDataKind = "own-data"

// holderOn is the module holding node-backup on a machine.
func holderOn(ctx context.Context, inv *inventory.Inventory, machine string) (string, error) {
	held, err := inv.Holdings(ctx)
	if err != nil {
		return "", err
	}
	for _, h := range held {
		if s, known := catalogue.SeatNamed(h.Claim); known && s.Name == catalogue.BackupSeat && h.Node == machine {
			return h.Module, nil
		}
	}
	return "", fmt.Errorf("nothing holds %s on %s, and it is the backup holder that deletes retired data there "+
		"(after a last restore point)", catalogue.BackupSeat, machine)
}

// deleteRetiredData has a machine's backup holder delete one retired item: never one declared now, and
// never one not retired. The holder takes a last restore point of it first, so the deletion can be
// undone until a person forgets that restore point; the record says deleted only once the holder says
// it is.
func deleteRetiredData(ctx context.Context, conn *nats.Conn, open *stores, r inventory.DataRecord, f handActFlags) error {
	inv := open.inventory
	if !r.Retired() {
		return fmt.Errorf("%s of %s on %s is not retired — only retired data is deleted. Nothing was done",
			r.Item, r.Module, r.Machine)
	}
	if r.Path == "" {
		return fmt.Errorf("%s of %s on %s was never measured, so where it is was never said; nothing was deleted",
			r.Item, r.Module, r.Machine)
	}
	if plan, _, err := planFor(ctx, open, r.Machine); err == nil {
		for _, m := range plan.Modules {
			if _, still := m.DataItem(r.Item); still && m.Module == r.Module {
				return fmt.Errorf("%s runs on %s again and declares %s: it is not retired any more. Nothing was done",
					r.Module, r.Machine, r.Item)
			}
		}
	}
	holder, err := holderOn(ctx, inv, r.Machine)
	if err != nil {
		return err
	}
	f.record(ctx, "cleanup delete", []string{r.Machine, r.Module, r.Item})
	args := map[string]any{"module": r.Module, "item": r.Item, "path": r.Path, "confirm": r.Item,
		"why": strings.TrimSpace(*f.why), "by": link.Caller(), "via": link.ViaController}
	ask := func(tool string) (deletion, error) {
		var d deletion
		answer, err := link.AskModuleToolOn(ctx, conn, holder, tool, r.Machine, args, 25*time.Second)
		if err != nil {
			return d, err
		}
		if answer.Error != "" {
			return d, fmt.Errorf("%s on %s refused: %s", holder, r.Machine, answer.Error)
		}
		return d, unmarshalAnswer(answer, &d)
	}
	d, err := ask(ToolDeleteRetired)
	if err != nil {
		return err
	}
	for waited := time.Duration(0); !d.Done && waited < deletionWait; waited += deletionPoll {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(deletionPoll):
		}
		if d, err = ask(ToolDeletedOutcome); err != nil {
			return err
		}
	}
	switch {
	case !d.Done:
		fmt.Printf("%s on %s is still taking the last restore point of %s and deleting it; `cleanup list` keeps "+
			"showing it until the holder says it is done — the same `cleanup delete` again reads how it went\n",
			holder, r.Machine, r.Path)
		return nil
	case !d.OK:
		return fmt.Errorf("%s on %s did NOT delete %s: %s", holder, r.Machine, r.Path, d.Error)
	}
	if err := inv.MarkDataDeleted(ctx, r.Machine, r.Module, r.Item, link.Caller(), strings.TrimSpace(*f.why), time.Now()); err != nil {
		return fmt.Errorf("%s deleted %s on %s, and it could not be recorded: %w", holder, r.Path, r.Machine, err)
	}
	fmt.Printf("%s on %s deleted %s of %s (%s, %s); its last restore point is %s, kept until a person forgets it\n",
		holder, r.Machine, r.Item, r.Module, r.Path, sizeWords(r.Size), orNever(d.Snapshot))
	return nil
}

// keptOnUnassign says, for an unassignment, the irreplaceable and valuable data each module leaves in its
// own directories on the machine:
// kept, and retired at the self-check's next run.
func keptOnUnassign(ctx context.Context, inv *inventory.Inventory, machine string, modules []string) []string {
	records, err := inv.Data(ctx)
	if err != nil {
		return []string{"what it leaves behind could not be read from the mesh's record: " + err.Error()}
	}
	var out []string
	for _, r := range records {
		if r.Machine != machine || r.DeletedAt != nil || !catalogue.Retires(r.Class) || !r.Owned {
			continue
		}
		for _, m := range modules {
			if r.Module == m {
				out = append(out, fmt.Sprintf("%s's %s (%s, %s) stays where it is: it is %s, so it is retired, "+
					"never removed — `cleanup list` shows it, `cleanup delete` alone removes it (novox/hq ADR 0233)",
					r.Module, r.Item, orUnknownPath(r.Path), sizeWords(r.Size), r.Class))
			}
		}
	}
	return out
}