package main

// What becomes of a consumer the mesh no longer asks for (novox/hq ADR 0230 — the operator's model,
// decided 2026-10-06; it replaces ADR 0229's withdrawal brake, which let one consumer go every hour).
//
// A consumer — a login, a client, a key, and the data behind it — is in one of three states:
//
//  1. ACTIVE.
//  2. RETIRED. The mesh stopped asking for it: its access is disabled, reversibly, and its login and
//     data are marked "to delete" with when and why. Nothing is deleted. Asked for again, the ordinary
//     create re-enables it at once, as it was.
//  3. DELETED, only through `cleanup delete`, a person's act, which this provider executes because it
//     owns its backend.
//
// **Stable removals.** A consumer is retired only once the same set of consumers has gone unasked BOTH
// in StablePasses (5) consecutive passes that read the file AND for at least StableFor (10 minutes)
// since the first pass that saw it. Five passes alone are twenty-five seconds — shorter than a
// controller restart, a store reconnecting or a file half written. A pass that could not read the file
// is not a result and starts both again; so does a different set. Additions and changes are never
// delayed.
//
// **Too many is a person.** A stable set of more than RetireAtOnce (3), or — where more than one is
// held — of more than RetireFraction (half) of those held, retires nothing: it WAITS, said in the
// journal and announced `provisioner.retirement` `waiting` (again every SayAgainEvery), which the
// controller raises as an urgent condition, until `retire approve <node> <module>` or `retire reject`.
// An unassignment the controller made itself is no exception: many changes at once means a person is
// at work, and a person confirms once. On 2026-10-04 one misread file withdrew seven consumers at once
// (issue 241); under this rule that is a question, not an act.
//
// **The backend remembers, not this process.** What is retired, since when and why is read from the
// backend's own mark (Adapter.Inventory), so a restart forgets nothing; a consumer the backend holds
// active that the mesh no longer asks for is retired by the same rules after a restart as before one.
// A consumer found disabled without the mark — withdrawn before this record — is ADOPTED as retired:
// marked, said, announced `adopted`, and its clock starts then.
//
// **A rejection lives as long as this process.** Rejected, the set is kept active and not asked about
// again while it stays the same set; a restart asks again — loudly, never silently.

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

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

// EventRetirement is the one event a provider says about retirement, its `change` saying what
// happened. The controller derives the permission to emit it for every module that receives
// contributions, as it does the standing events (novox/hq ADR 0224, 0230).
const EventRetirement = "provisioner.retirement"

// The changes EventRetirement carries.
const (
	ChangeWaiting   = "waiting"   // a stable set over the bound waits for a person
	ChangeSettled   = "settled"   // a waiting or rejected set ended without being retired
	ChangeApproved  = "approved"  // a person approved the waiting (or rejected) set
	ChangeRejected  = "rejected"  // a person kept the waiting set active
	ChangeRetired   = "retired"   // consumers disabled and marked to delete
	ChangeReenabled = "reenabled" // a retired consumer asked for again, enabled as it was
	ChangeDeleted   = "deleted"   // a retired consumer deleted, by a person
	ChangeAdopted   = "adopted"   // found disabled without the mark, now retired on record
)

// StableFor is the least time the same unasked set must hold before it is retired (novox/hq ADR 0230):
// longer than a controller restart, a store reconnecting or a file half written.
const StableFor = 10 * time.Minute

// KindConsumer is a retired consumer. A backend may list other things set aside for deletion under a
// word of its own (postgres: a database renamed aside).
const KindConsumer = "consumer"

// Retired is one thing a provider holds retired, as its backend's mark says.
type Retired struct {
	Consumer string
	// Node is the consumer's machine, where the mark records it.
	Node string
	// RetiredAt is when it was retired; zero for one found disabled without the mesh's mark.
	RetiredAt time.Time
	Why       string
	// SizeBytes is what deleting it would free; -1 when the backend cannot say.
	SizeBytes int64
	Kind      string
}

// Inventory is what a backend holds that the mesh made.
type Inventory struct {
	Active  []string
	Retired []Retired
	// Sizes is what each active consumer keeps on the backend, in bytes, where the backend can say: what
	// the controller tells an empty replacement of a consumer's data by (novox/hq ADR 0233).
	Sizes map[string]int64
}

// retirement is the loop's memory of the sets it is counting, waiting on and was told to keep.
type retirement struct {
	key       string // the unasked set, joined
	count     int    // consecutive passes that saw it
	firstSeen time.Time
	waiting   *waitingSet
	rejected  *rejectedSet
	retired   map[string]retiredEntry
	lastWant  map[string]bool
	seeded    bool
	seedSaid  string
}

type waitingSet struct {
	set    []string
	since  time.Time
	saidAt time.Time
	held   int
}

type rejectedSet struct {
	set     []string
	by, why string
	at      time.Time
}

type retiredEntry struct {
	node string
	at   time.Time
	why  string
}

// seed reads what the backend holds, once per process: an active consumer the mesh no longer asks
// for is held, so it is retired by the same rules as one this process made; one found disabled
// without the mark is adopted as retired.
func (h *Harness) seed(ctx context.Context, want map[string]bool) {
	if h.r.seeded {
		return
	}
	inv, err := h.Adapter.Inventory(ctx)
	if err != nil {
		if said := err.Error(); said != h.r.seedSaid {
			h.say("cannot list what the backend holds (%v); a consumer the mesh stopped asking for while this "+
				"provider was down is not retired until it can", err)
			h.r.seedSaid = said
		}
		return
	}
	h.r.seeded = true
	for _, as := range inv.Active {
		if _, held := h.applied[as]; !held && !want[as] {
			// No hash: asked for again, it is applied again, which is harmless and marks it.
			h.applied[as] = appliedEntry{}
			h.say("%s: held by the backend and no longer asked for; retired once that holds for %d passes "+
				"and %s (novox/hq ADR 0230)", as, h.StablePasses, h.StableFor)
		}
	}
	now := h.Now()
	for _, r := range inv.Retired {
		if r.Kind != "" && r.Kind != KindConsumer {
			continue
		}
		if !r.RetiredAt.IsZero() {
			h.r.retired[r.Consumer] = retiredEntry{node: r.Node, at: r.RetiredAt, why: r.Why}
			continue
		}
		if want[r.Consumer] {
			continue // asked for: this pass's create enables it
		}
		why := "found disabled without the mesh's mark — withdrawn before retirement was recorded " +
			"(novox/hq ADR 0230); retired as found"
		if err := h.Adapter.Retire(ctx, r.Consumer, nil, why, now); err != nil {
			h.say("%s: found disabled and could not be marked retired, will try at the next start: %v", r.Consumer, err)
			continue
		}
		h.r.retired[r.Consumer] = retiredEntry{node: r.Node, at: now, why: why}
		h.say("%s: ADOPTED as retired — %s", r.Consumer, why)
		h.announce(EventRetirement, h.retirementBody(ChangeAdopted, []map[string]any{
			{"consumer": r.Consumer, "node": r.Node, "retired_at": stamp(now), "why": why, "size_bytes": r.SizeBytes},
		}, map[string]any{"why": why}))
	}
}

// retireUnasked counts the passes that saw the same set no longer asked for and, once it is stable,
// retires it — or, past the bound, waits for a person.
func (h *Harness) retireUnasked(ctx context.Context, want map[string]bool) {
	var unasked []string
	for as := range h.applied {
		if !want[as] {
			unasked = append(unasked, as)
		}
	}
	sort.Strings(unasked)
	key := strings.Join(unasked, "\x00")
	now := h.Now()
	switch {
	case len(unasked) == 0:
		h.settle("every consumer held is asked for again")
		h.r.key, h.r.count = "", 0
		return
	case key != h.r.key:
		if h.r.waiting != nil || h.r.rejected != nil {
			h.settle("the set the mesh no longer asks for changed")
		}
		h.r.key, h.r.count, h.r.firstSeen = key, 1, now
		h.say("no longer asked for: %s; retired once the same set holds for %d passes and %s",
			strings.Join(unasked, ", "), h.StablePasses, h.StableFor)
	default:
		h.r.count++
	}
	if h.r.rejected != nil || h.r.count < h.StablePasses || now.Sub(h.r.firstSeen) < h.StableFor {
		return
	}
	if h.overTheBound(len(unasked), len(h.applied)) {
		h.wait(unasked)
		return
	}
	why := fmt.Sprintf("the mesh stopped asking for it: the same result in %d consecutive passes over %s, from %s",
		h.r.count, now.Sub(h.r.firstSeen).Round(time.Second), stamp(h.r.firstSeen))
	h.retire(ctx, unasked, why, nil)
}

// overTheBound says retiring n of the held consumers at once needs a person.
func (h *Harness) overTheBound(n, held int) bool {
	if n == 0 {
		return false
	}
	return n > h.RetireAtOnce || (held > 1 && float64(n) > h.RetireFraction*float64(held))
}

func (h *Harness) bound(held int) string {
	return fmt.Sprintf("more than %d at once, or more than %.0f%% of the %d held", h.RetireAtOnce,
		h.RetireFraction*100, held)
}

// wait holds a stable set over the bound for a person, said once and announced again every
// SayAgainEvery so a controller that missed the first hears the next.
func (h *Harness) wait(set []string) {
	now := h.Now()
	w := h.r.waiting
	if w == nil {
		w = &waitingSet{set: set, since: now, held: len(h.applied)}
		h.r.waiting = w
		h.say("RETIREMENT WAITS FOR A PERSON: the mesh no longer asks for %d of the %d consumer(s) this provider "+
			"holds (%s), %s. Nothing is retired; each stays active until `retire approve %s <module>` or "+
			"`retire reject` (novox/hq ADR 0230)", len(set), w.held, strings.Join(set, ", "), h.bound(w.held), h.Node)
	} else if now.Sub(w.saidAt) < h.SayAgainEvery {
		return
	}
	w.saidAt = now
	h.announce(EventRetirement, h.retirementBody(ChangeWaiting, h.named(set), map[string]any{
		"held": w.held, "bound": h.bound(w.held), "since": stamp(w.since),
	}))
}

// settle ends a waiting or rejected set that the mesh's own answer made moot.
func (h *Harness) settle(why string) {
	var set []string
	switch {
	case h.r.waiting != nil:
		set = h.r.waiting.set
	case h.r.rejected != nil:
		set = h.r.rejected.set
	default:
		return
	}
	h.r.waiting, h.r.rejected = nil, nil
	h.say("retirement of %s settled without retiring: %s", strings.Join(set, ", "), why)
	h.announce(EventRetirement, h.retirementBody(ChangeSettled, h.named(set), map[string]any{"why": why}))
}

// retire disables and marks each consumer of set; one that fails stays held and is tried again once
// its set is stable again.
func (h *Harness) retire(ctx context.Context, set []string, why string, extra map[string]any) []string {
	now := h.Now()
	held := len(h.applied)
	var done []string
	var said []map[string]any
	for _, as := range set {
		was, ok := h.applied[as]
		if !ok {
			continue
		}
		if err := h.Adapter.Retire(ctx, as, was.derived, why, now); err != nil {
			h.say("%s: retiring failed, will try again: %v", as, err)
			continue
		}
		delete(h.applied, as)
		delete(h.lost, as)
		delete(h.failing, as)
		h.r.retired[as] = retiredEntry{node: was.node, at: now, why: why}
		h.recovered(as, "retired")
		h.say("%s: RETIRED — its access disabled and its data kept, marked to delete (%s). Asked for again "+
			"it is re-enabled as it was; it is deleted only by `cleanup delete` (novox/hq ADR 0230)", as, why)
		done = append(done, as)
		said = append(said, map[string]any{"consumer": as, "node": was.node, "retired_at": stamp(now), "why": why})
	}
	h.r.key, h.r.count, h.r.waiting, h.r.rejected = "", 0, nil, nil
	if len(done) > 0 {
		body := map[string]any{"held": held, "why": why}
		for k, v := range extra {
			body[k] = v
		}
		h.announce(EventRetirement, h.retirementBody(ChangeRetired, said, body))
	}
	return done
}

// reenabled says a retired consumer was asked for again and its create enabled it.
func (h *Harness) reenabled(as, node string) {
	e, was := h.r.retired[as]
	if !was {
		return
	}
	delete(h.r.retired, as)
	h.say("%s: asked for again — re-enabled as it was, its mark to delete cleared (retired %s: %s)", as,
		stamp(e.at), e.why)
	h.announce(EventRetirement, h.retirementBody(ChangeReenabled, []map[string]any{
		{"consumer": as, "node": node, "retired_at": stamp(e.at), "why": e.why},
	}, nil))
}

// Approve retires exactly the set waiting (or the set rejected) — a person's confirmation.
func (h *Harness) Approve(ctx context.Context, consumers []string, why, by, via string) ([]string, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	if strings.TrimSpace(why) == "" {
		return nil, errors.New("approve: say why")
	}
	set := sorted(consumers)
	var held []string
	switch {
	case h.r.waiting != nil && same(set, h.r.waiting.set):
		held = h.r.waiting.set
	case h.r.rejected != nil && same(set, h.r.rejected.set):
		held = h.r.rejected.set
	default:
		return nil, fmt.Errorf("approve: %s is not the set waiting here — %s", orNone(set), h.waitingWords())
	}
	h.say("retirement of %s APPROVED by %s: %s", strings.Join(held, ", "), orSomebody(by), why)
	h.announce(EventRetirement, h.retirementBody(ChangeApproved, h.named(held),
		map[string]any{"why": why, "by": by, "via": via}))
	reason := fmt.Sprintf("the mesh stopped asking for it; approved by %s: %s", orSomebody(by), why)
	return h.retire(ctx, held, reason, map[string]any{"by": by, "via": via}), nil
}

// Reject keeps the waiting set active; it is not asked about again while it stays the same set.
func (h *Harness) Reject(consumers []string, why, by, via string) ([]string, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	if strings.TrimSpace(why) == "" {
		return nil, errors.New("reject: say why")
	}
	set := sorted(consumers)
	if h.r.waiting == nil || !same(set, h.r.waiting.set) {
		return nil, fmt.Errorf("reject: %s is not the set waiting here — %s", orNone(set), h.waitingWords())
	}
	h.r.rejected = &rejectedSet{set: h.r.waiting.set, by: by, why: why, at: h.Now()}
	h.r.waiting = nil
	h.say("retirement of %s REJECTED by %s: %s. Kept active, and not asked about again while the mesh goes on "+
		"not asking for exactly these (until this provider restarts)", strings.Join(set, ", "), orSomebody(by), why)
	h.announce(EventRetirement, h.retirementBody(ChangeRejected, h.named(set),
		map[string]any{"why": why, "by": by, "via": via}))
	return set, nil
}

// DeleteRetired deletes one retired consumer, by a person's word: never an active one, never one the
// mesh asks for.
func (h *Harness) DeleteRetired(ctx context.Context, consumer, why, by, via string) (int64, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	if strings.TrimSpace(why) == "" {
		return 0, errors.New("delete: say why")
	}
	if h.r.lastWant[consumer] {
		return 0, fmt.Errorf("delete: the mesh asks for %s — it is not retired", consumer)
	}
	if _, held := h.applied[consumer]; held {
		return 0, fmt.Errorf("delete: %s is active here — only a retired consumer is deleted", consumer)
	}
	inv, err := h.Adapter.Inventory(ctx)
	if err != nil {
		return 0, fmt.Errorf("delete: what the backend holds cannot be read, so nothing is deleted: %w", err)
	}
	var found *Retired
	for i := range inv.Retired {
		if inv.Retired[i].Consumer == consumer {
			found = &inv.Retired[i]
		}
	}
	if found == nil {
		return 0, fmt.Errorf("delete: %s is not retired here — only a retired consumer is deleted", consumer)
	}
	freed, err := h.Adapter.Delete(ctx, *found)
	if err != nil {
		return 0, err
	}
	delete(h.r.retired, consumer)
	h.say("%s: DELETED by %s: %s (retired %s: %s; %d bytes freed)", consumer, orSomebody(by), why,
		stampOr(found.RetiredAt), found.Why, freed)
	h.announce(EventRetirement, h.retirementBody(ChangeDeleted, []map[string]any{{
		"consumer": consumer, "node": found.Node, "retired_at": stampOr(found.RetiredAt), "why": found.Why,
		"size_bytes": found.SizeBytes, "kind": kindOf(*found),
	}}, map[string]any{"why": why, "by": by, "via": via, "freed_bytes": freed}))
	return freed, nil
}

// Retirement is what a person (and the controller's `cleanup list` and its probe) asks: what is
// held, waiting, rejected and retired here.
func (h *Harness) Retirement(ctx context.Context) (map[string]any, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	inv, err := h.Adapter.Inventory(ctx)
	if err != nil {
		return nil, err
	}
	var held []string
	for as := range h.applied {
		held = append(held, as)
	}
	sort.Strings(held)
	retired := []map[string]any{}
	for _, r := range inv.Retired {
		retired = append(retired, map[string]any{"consumer": r.Consumer, "node": r.Node,
			"retired_at": stampOr(r.RetiredAt), "why": r.Why, "size_bytes": r.SizeBytes, "kind": kindOf(r)})
	}
	sizes := map[string]int64{}
	for _, as := range held {
		if size, ok := inv.Sizes[as]; ok && size >= 0 {
			sizes[as] = size
		}
	}
	out := map[string]any{"resource": h.Resource, "node": h.Node, "held": orEmptyList(held), "held_sizes": sizes,
		"stable_passes": h.StablePasses, "stable_for_seconds": int(h.StableFor.Seconds()), "bound": h.bound(len(h.applied)), "waiting": nil, "rejected": nil,
		"retired": retired}
	if w := h.r.waiting; w != nil {
		out["waiting"] = map[string]any{"consumers": h.named(w.set), "since": stamp(w.since), "held": w.held}
	}
	if r := h.r.rejected; r != nil {
		out["rejected"] = map[string]any{"consumers": h.named(r.set), "by": r.by, "why": r.why, "at": stamp(r.at)}
	}
	return out, nil
}

func (h *Harness) waitingWords() string {
	switch {
	case h.r.waiting != nil:
		return "waiting: " + strings.Join(h.r.waiting.set, ", ")
	case h.r.rejected != nil:
		return "nothing waits; rejected and kept: " + strings.Join(h.r.rejected.set, ", ")
	}
	return "nothing waits for a person here"
}

// named is a set with each consumer's machine, as the events and the tools say it.
func (h *Harness) named(set []string) []map[string]any {
	out := make([]map[string]any, 0, len(set))
	for _, as := range set {
		out = append(out, map[string]any{"consumer": as, "node": h.applied[as].node})
	}
	return out
}

func (h *Harness) retirementBody(change string, consumers []map[string]any, extra map[string]any) map[string]any {
	body := map[string]any{"provider": h.Resource, "provider-node": h.Node, "change": change,
		"consumers": consumers, "at": stamp(h.Now())}
	for k, v := range extra {
		body[k] = v
	}
	return body
}

// RetirementTools are the four tools every provider serves for retirement, the same names in every
// module: the controller's `retire` and `cleanup` verbs ask them on the provider's machine.
func RetirementTools(h *Harness) []stdio.Tool {
	if h == nil {
		return nil
	}
	ask := func() (context.Context, context.CancelFunc) {
		return context.WithTimeout(context.Background(), 2*time.Minute)
	}
	who := map[string]any{
		"why": map[string]any{"type": "string", "description": "why — required, kept in the hand-act log"},
		"by":  map[string]any{"type": "string", "description": "who decided"},
		"via": map[string]any{"type": "string", "description": "mesh-controller when asked through its verbs"},
	}
	withSet := map[string]any{"consumers": map[string]any{"type": "array", "items": map[string]any{"type": "string"},
		"description": "the set, exactly as provisioner_retirement names it"}}
	for k, v := range who {
		withSet[k] = v
	}
	withOne := map[string]any{
		"consumer": map[string]any{"type": "string", "description": "the retired consumer to delete"},
		"confirm":  map[string]any{"type": "string", "description": "the consumer's name again, to say this is meant"},
	}
	for k, v := range who {
		withOne[k] = v
	}
	return []stdio.Tool{
		{Name: "provisioner_retirement",
			Description: "What this provider holds, what waits for a person to approve its retirement, what was rejected, " +
				"and every retired consumer with when, why and its size (novox/hq ADR 0230).",
			Run: func(map[string]any) (any, error) {
				ctx, cancel := ask()
				defer cancel()
				return h.Retirement(ctx)
			}},
		{Name: "provisioner_retire_approve",
			Description: "Retire exactly the set waiting for a person (or the set rejected earlier): access disabled, data " +
				"kept and marked to delete. Use the controller's `retire approve`, which records the hand act.",
			Input: withSet,
			Run: func(args map[string]any) (any, error) {
				ctx, cancel := ask()
				defer cancel()
				done, err := h.Approve(ctx, strs(args["consumers"]), argString(args, "why"), argString(args, "by"), argString(args, "via"))
				if err != nil {
					return nil, err
				}
				return map[string]any{"retired": orEmptyList(done)}, nil
			}},
		{Name: "provisioner_retire_reject",
			Description: "Keep the set waiting for a person active. Use the controller's `retire reject`.",
			Input:       withSet,
			Run: func(args map[string]any) (any, error) {
				kept, err := h.Reject(strs(args["consumers"]), argString(args, "why"), argString(args, "by"), argString(args, "via"))
				if err != nil {
					return nil, err
				}
				return map[string]any{"kept": kept}, nil
			}},
		{Name: "provisioner_delete",
			Description: "Delete one RETIRED consumer — its login and its data — for good. Refused for anything active or " +
				"asked for. Use the controller's `cleanup delete`, which records the hand act.",
			Input: withOne,
			Run: func(args map[string]any) (any, error) {
				consumer := argString(args, "consumer")
				if consumer == "" || argString(args, "confirm") != consumer {
					return nil, errors.New("delete: name the consumer, and repeat it in confirm")
				}
				ctx, cancel := ask()
				defer cancel()
				freed, err := h.DeleteRetired(ctx, consumer, argString(args, "why"), argString(args, "by"), argString(args, "via"))
				if err != nil {
					return nil, err
				}
				return map[string]any{"deleted": consumer, "freed_bytes": freed}, nil
			}},
	}
}

func strs(v any) []string {
	list, _ := v.([]any)
	out := []string{}
	for _, x := range list {
		if s, ok := x.(string); ok && s != "" {
			out = append(out, s)
		}
	}
	return out
}

func sorted(list []string) []string {
	out := append([]string(nil), list...)
	sort.Strings(out)
	return out
}

func same(a, b []string) bool {
	return strings.Join(sorted(a), "\x00") == strings.Join(sorted(b), "\x00")
}

func orNone(set []string) string {
	if len(set) == 0 {
		return "an empty set"
	}
	return strings.Join(set, ", ")
}

func orSomebody(by string) string {
	if by == "" {
		return "a person"
	}
	return by
}

func orEmptyList(list []string) []string {
	if list == nil {
		return []string{}
	}
	return list
}

func kindOf(r Retired) string {
	if r.Kind == "" {
		return KindConsumer
	}
	return r.Kind
}

func stamp(t time.Time) string { return t.UTC().Format(time.RFC3339) }

func stampOr(t time.Time) string {
	if t.IsZero() {
		return ""
	}
	return stamp(t)
}

func argString(args map[string]any, key string) string {
	s, _ := args[key].(string)
	return s
}