package main

// The asker on the bus: its asks in the controller's bucket `asked`, its asks and cancels published on the
// seat under the controller's name, the verbs a warrant chooses called with the controller's grant, and the
// router's record of its asks read under its name (novox/hq ADR 0259).

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

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

	"git.novox.be/novox/mesh-sdk/go/asks"

	"github.com/novox/mesh-controller/internal/broker"
	"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"
)

// askerFrom is the serving controller's asker; nil in any other process.
var askerFrom *asker

// askWithin is how long a verb a warrant chose is given to answer.
const askWithin = time.Minute

type busAsked struct{ conn *nats.Conn }

func (b busAsked) kv(ctx context.Context) (jetstream.KeyValue, error) {
	js, err := jetstream.New(b.conn)
	if err != nil {
		return nil, err
	}
	return js.KeyValue(ctx, broker.AskedBucket)
}

func (b busAsked) Get(ctx context.Context, id string) (*asked, error) {
	kv, err := b.kv(ctx)
	if err != nil {
		return nil, err
	}
	e, err := kv.Get(ctx, id)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var r asked
	return &r, json.Unmarshal(e.Value(), &r)
}

// Create keeps a new ask under its id, and only where none is kept: never over another.
func (b busAsked) Create(ctx context.Context, r asked) error {
	kv, err := b.kv(ctx)
	if err != nil {
		return err
	}
	body, err := json.Marshal(r)
	if err != nil {
		return err
	}
	_, err = kv.Create(ctx, r.ID, body)
	return err
}

// Change applies change to the ask kept under id by compare-and-set on its key's revision (the review of
// 2026-10-09, L2): read, changed, and written only over the revision read; when another write came between,
// read again and asked again, at most askChangeTries times. change says whether to write at all.
func (b busAsked) Change(ctx context.Context, id string, change func(*asked) bool) (bool, error) {
	kv, err := b.kv(ctx)
	if err != nil {
		return false, err
	}
	for try := 0; try < askChangeTries; try++ {
		e, err := kv.Get(ctx, id)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		var r asked
		if err := json.Unmarshal(e.Value(), &r); err != nil {
			return false, err
		}
		if !change(&r) {
			return false, nil
		}
		body, err := json.Marshal(r)
		if err != nil {
			return false, err
		}
		if _, err := kv.Update(ctx, id, body, e.Revision()); err != nil {
			var api *jetstream.APIError
			if errors.Is(err, jetstream.ErrKeyExists) || (errors.As(err, &api) && api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence) {
				continue
			}
			return false, err
		}
		return true, nil
	}
	return false, fmt.Errorf("the ask %s changed under every one of %d tries", id, askChangeTries)
}

func (b busAsked) All(ctx context.Context) ([]asked, error) {
	kv, err := b.kv(ctx)
	if err != nil {
		return nil, err
	}
	lister, err := kv.ListKeys(ctx)
	if err != nil {
		return nil, err
	}
	defer func() { _ = lister.Stop() }()
	var out []asked
	for k := range lister.Keys() {
		e, err := kv.Get(ctx, k)
		if err != nil {
			continue
		}
		var r asked
		if json.Unmarshal(e.Value(), &r) == nil {
			out = append(out, r)
		}
	}
	return out, nil
}

// callAction performs an action's verb as the controller, through the grant that names it.
func callAction(conn *nats.Conn) func(ctx context.Context, a conditions.Action, args map[string]string) error {
	return func(ctx context.Context, a conditions.Action, args map[string]string) error {
		seat, verb, ok := strings.Cut(a.Verb, ".")
		if !ok {
			return fmt.Errorf("%q names no seat and verb", a.Verb)
		}
		body := map[string]any{}
		for k, v := range args {
			body[k] = v
		}
		if seat == catalogue.DeliverySeat {
			_, err := askDeliveryOwner(ctx, conn, verb, body)
			return err
		}
		granted := false
		for _, v := range broker.VerbsTheControllerActsOnAWarrant {
			granted = granted || (v.Seat == seat && v.Verb == verb)
		}
		if !granted {
			return fmt.Errorf("%s: %w", a.Verb, errNotGranted)
		}
		var answer link.Answer
		var err error
		if a.Machine != "" {
			answer, err = link.AskSeatTool(ctx, conn, seat, verb, a.Machine, body, askWithin)
		} else {
			answer, err = link.AskMeshSeatTool(ctx, conn, seat, verb, body, askWithin)
		}
		if err != nil {
			return err
		}
		if answer.Error != "" {
			return fmt.Errorf("%s refused: %s", a.Verb, answer.Error)
		}
		return nil
	}
}

// routerRecordOf reads the router's record of one of the controller's asks, under its name, and answers
// how it ended when it did: the bucket is the one the asks seat's declarer names as its records.
func routerRecordOf(conn *nats.Conn, inv *inventory.Inventory) func(ctx context.Context, id string) (*asks.Warrant, error) {
	return func(ctx context.Context, id string) (*asks.Warrant, error) {
		bucket, err := asksRecords(ctx, inv)
		if err != nil || bucket == "" {
			return nil, err
		}
		reply, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+askerName+"."+id, nil)
		if err != nil {
			return nil, err
		}
		if reply.Header.Get("Status") != "" {
			return nil, nil // none, or not readable: the event says it
		}
		var rec struct {
			State   string        `json:"state"`
			Warrant *asks.Warrant `json:"warrant"`
		}
		if json.Unmarshal(reply.Data, &rec) != nil || rec.State == "open" || rec.Warrant == nil {
			return nil, nil
		}
		return rec.Warrant, nil
	}
}

// asksRecords is the bucket the asks seat's declarer keeps its record of asks in.
func asksRecords(ctx context.Context, inv *inventory.Inventory) (string, error) {
	declared, err := inv.Catalogue(ctx)
	if err != nil {
		return "", err
	}
	for _, m := range declared {
		for _, s := range m.DefinesSeats {
			if s.Name == broker.AsksSeat && len(s.Records) > 0 {
				return broker.BucketName(m.Module, s.Records[0]), nil
			}
		}
	}
	return "", nil
}

// routerHereIn says whether a module declaring the asks seat, with its ask named by its caller, is assigned:
// without it nothing takes an ask, and asking would only fill a queue nobody reads.
func routerHereIn(inv *inventory.Inventory) func(ctx context.Context) (bool, error) {
	return func(ctx context.Context) (bool, error) {
		entries, err := inv.Catalogued(ctx)
		if err != nil {
			return false, err
		}
		for _, e := range entries {
			for _, s := range e.Manifest.DefinesSeats {
				if s.Name == broker.AsksSeat && s.NamedByCaller("ask") && len(e.On) > 0 {
					return true, nil
				}
			}
		}
		return false, nil
	}
}

// grantHeldIn says whether the bus holds the controller's grant to ask (novox/hq issue 353), judged on the two
// facts that make it so, and failing closed on each:
//
//  1. **composed**: the user list the mesh composes now grants the controller itself to publish its ask
//     (controllerMayAsk) — the controller's own grant, not whether any user of the list changed;
//  2. **in force**: the bus's own module, on the machine holding the bus, says the user list there allows it
//     and the server reloaded after that list was written (the nats module's `nats_user_can`, `in_force`) — a
//     list sent is not a list the bus enforces until it reloads.
//
// Anything that cannot be worked out is not held, with why, so the conditions that need the operator are said
// as undelivered rather than asked into a refusal.
func grantHeldIn(open *stores, conn *nats.Conn) func(ctx context.Context) (bool, string, error) {
	return grantHeldBy(composedGrantIn(open), busGrantIn(open, conn), time.Now)
}

// grantJudge is one of the two facts: whether it holds, and why not.
type grantJudge func(ctx context.Context) (bool, string, error)

// grantHeldBy is the judgement over the two facts, remembered for grantLookEvery: composing the list resolves
// the bus's machine whole, and asking the bus's module is a call.
func grantHeldBy(composed, inForce grantJudge, now func() time.Time) func(ctx context.Context) (bool, string, error) {
	var mu sync.Mutex
	var at time.Time
	var held bool
	var why string
	return func(ctx context.Context) (bool, string, error) {
		mu.Lock()
		defer mu.Unlock()
		if !at.IsZero() && now().Sub(at) < grantLookEvery {
			return held, why, nil
		}
		held, why = judgeGrant(ctx, composed, inForce)
		at = now()
		return held, why, nil
	}
}

// judgeGrant is held only when both facts hold; an error is a fact not known, never a yes. A fact that holds
// with a caveat (a reload not confirmed) gives that caveat as the why of a yes.
func judgeGrant(ctx context.Context, composed, inForce grantJudge) (bool, string) {
	var caveat []string
	for _, fact := range []struct {
		judge grantJudge
		what  string
	}{{composed, "whether the mesh composes the controller a grant to ask"}, {inForce, "whether the bus enforces it"}} {
		if fact.judge == nil {
			return false, fact.what + " is not judged here"
		}
		ok, why, err := fact.judge(ctx)
		if err != nil {
			return false, fact.what + " could not be worked out: " + err.Error()
		}
		if !ok {
			return false, why
		}
		if why != "" {
			caveat = append(caveat, why)
		}
	}
	return true, strings.Join(caveat, "; ")
}

// controllerMayAsk says whether the controller's own grant, as composed from these records, lets it publish its
// ask (ADR 0259 §3: composed while a router that takes asks under its asker's name is assigned).
func controllerMayAsk(records broker.Records) (bool, error) {
	users, err := broker.Users(records)
	if err != nil {
		return false, err
	}
	for _, u := range users {
		if u.Kind != broker.KindController {
			continue
		}
		perms, err := broker.PermissionsFor(u)
		if err != nil {
			return false, err
		}
		return broker.MayPublish(perms, asks.AskSubject(askerName)), nil
	}
	return false, errors.New("the composition holds no controller")
}

// composedGrantIn is the first fact, from the mesh's records.
func composedGrantIn(open *stores) grantJudge {
	return func(ctx context.Context) (bool, string, error) {
		if open == nil || open.inventory == nil {
			return false, "", errors.New("the mesh's records are not open")
		}
		records, err := open.inventory.BusRecords(ctx)
		if err != nil {
			return false, "", err
		}
		may, err := controllerMayAsk(records)
		if err != nil || may {
			return may, "", err
		}
		return false, "the mesh composes the controller no grant to ask: no router that takes asks under its " +
			"asker's name is assigned", nil
	}
}

// busGrantIn is the second fact, asked of the bus's own module on the machine holding the bus, with the check
// main made before it (novox/hq issue 353, PR 186: is the user list sent to the bus's machine the one composed
// now) kept for the cases the module's word does not settle.
func busGrantIn(open *stores, conn *nats.Conn) grantJudge {
	ask := func(ctx context.Context) (string, json.RawMessage, error) {
		if open == nil || open.inventory == nil || conn == nil {
			return "", nil, errors.New("the mesh's records or the bus are not open")
		}
		holders, err := seatHolders(ctx, open.inventory)
		if err != nil {
			return "", nil, err
		}
		h, held := holders[theBrokerSeat]
		if !held || h.Node == "" || h.Module == "" {
			return "", nil, nil
		}
		a, err := link.AskModuleToolOn(ctx, conn, h.Module, busUserCanTool, h.Node, map[string]any{
			"user": broker.ControllerName, "action": "publish", "subject": asks.AskSubject(askerName)}, 15*time.Second)
		if err != nil {
			return h.Node, nil, err
		}
		if a.Error != "" {
			return h.Node, nil, errors.New(a.Error)
		}
		return h.Node, a.Result, nil
	}
	list := func(ctx context.Context) (busListState, error) {
		if open == nil || open.inventory == nil {
			return busListState{}, errors.New("the mesh's records are not open")
		}
		s, err := brokerListState(ctx, open, nil)
		if err != nil || s.Machine == "" || !s.Behind {
			return s, err
		}
		step, err := busHeld(ctx, open.inventory, []string{s.Machine})
		if err != nil {
			return busListState{}, err
		}
		s.BusStep = step[s.Machine]
		return s, nil
	}
	return busGrantBy(ask, list)
}

// busListState is the user list on the bus's machine as the controller's record knows it: whether the list
// sent there is behind the one composed now, why it could not be composed, and the bus step a send there waits
// on (that machine cannot be pushed until a person's `bus upgrade`).
type busListState struct {
	Machine    string
	Behind     bool
	Uncomposed error
	BusStep    string
}

// remedy is the one thing that sends the bus's machine the list composed now: the plan's own reason when it
// cannot be composed, the bus upgrade while sends there wait on it, and otherwise a push of that machine.
func (s busListState) remedy(node string) string {
	if s.Machine != "" {
		node = s.Machine
	}
	switch {
	case s.Uncomposed != nil:
		return fmt.Sprintf("the user list for %s cannot be composed (%v); `plan %s` says what to fix first",
			node, s.Uncomposed, node)
	case s.BusStep != "":
		return fmt.Sprintf("sends to %s wait on the bus's planned step, so a push will not carry it: %s", node, s.BusStep)
	}
	return fmt.Sprintf("`push %s` carries it", node)
}

// busGrantBy judges the bus's word, and names for each way it is not held the one thing that makes it held:
//
//   - **in force**: held;
//   - **allowed, from a module older than `in_force`**: held, with the caveat that the reload is not confirmed.
//     The file allows the publish, and the bus reloads within about 5 s of a change to it, so the worst case is
//     one ask refused in those seconds, which is asked again (issue 369). The list check is not consulted:
//     while a bus step is pending the bus's machine cannot be pushed, so any change to the composed list would
//     read as behind for good and silence every ask until a person's bus upgrade (the review of PR 188);
//   - **allowed, written, not reloaded** (the module gives both times): wait about 5 s;
//   - **allowed, in force not known** (the module could not read a time): what is not known, said;
//   - **not allowed, or the module cannot answer**: the list check decides — the remedy that sends the list
//     when it is behind (busListState.remedy), and, for a module that cannot answer, held while the list sent
//     is current, as main judged it.
//
// ask answers the machine holding the bus and its module's answer ("" when nothing holds the bus); list is the
// record's word on the list sent to that machine.
func busGrantBy(ask func(ctx context.Context) (string, json.RawMessage, error),
	list func(ctx context.Context) (busListState, error)) grantJudge {
	return func(ctx context.Context) (bool, string, error) {
		node, raw, askErr := ask(ctx)
		if askErr == nil && node == "" {
			return false, "no module holds the bus, so nothing says what it enforces", nil
		}
		var answer busAnswer
		if askErr == nil {
			if answer, askErr = readBusAnswer(raw); askErr != nil {
				askErr = fmt.Errorf("its answer could not be read: %w", askErr)
			}
		}
		if askErr != nil {
			if node == "" {
				node = "the bus's machine"
			}
			s, err := list(ctx)
			if err != nil {
				return false, "", fmt.Errorf("the bus module on %s could not answer (%v), and whether the list sent "+
					"there is the one composed now could not be worked out: %w", node, askErr, err)
			}
			if s.Behind {
				return false, fmt.Sprintf("the bus module on %s could not answer (%v), and the user list sent there is "+
					"behind what the mesh composes; %s", node, askErr, s.remedy(node)), nil
			}
			return true, fmt.Sprintf("the bus module on %s could not answer (%v); judged by the user list sent there, "+
				"which is the one composed now", node, askErr), nil
		}
		notAllowed := func(older string) (bool, string, error) {
			s, err := list(ctx)
			if err != nil {
				return false, "", fmt.Errorf("the user list on %s does not allow the controller to ask%s, and whether "+
					"the list sent there is the one composed now could not be worked out: %w", node, older, err)
			}
			if s.Behind {
				return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s: it is behind "+
					"what the mesh composes; %s", node, older, s.remedy(node)), nil
			}
			return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s, though the list "+
				"sent there is the one composed now; %s again", node, older, s.remedy(node)), nil
		}
		switch {
		case answer.InForce == nil && answer.Allowed:
			return true, fmt.Sprintf("the user list on %s allows the controller to ask, but its reload is not "+
				"confirmed: the bus module there is older than the answer whether a grant is in force, and the bus "+
				"upgrade (`bus upgrade`) carries one that confirms it", node), nil
		case answer.InForce == nil:
			return notAllowed(" (the bus module there is older than the answer whether a grant is in force)")
		case !answer.Allowed:
			return notAllowed("")
		case *answer.InForce:
			return true, "", nil
		case answer.FileWritten != "" && answer.ServerLoaded != "":
			return false, fmt.Sprintf("the bus on %s has the controller's grant to ask written (%s) but has not "+
				"reloaded since (last load %s); it reloads in place within about 5 s, so nothing is to be done but wait",
				node, answer.FileWritten, answer.ServerLoaded), nil
		}
		why := answer.Why
		if why == "" {
			why = "it gives no reason"
		}
		return false, fmt.Sprintf("the user list on %s allows the controller to ask, but whether the bus enforces it "+
			"is not known: %s", node, why), nil
	}
}

// busUserCanTool is the bus module's answer to whether a user may do something, and whether it is in force.
const busUserCanTool = "nats_user_can"

// busAnswer is the bus module's answer to busUserCanTool; InForce is nil from a module older than that field,
// and FileWritten and ServerLoaded are given only when it could read both times (mesh-catalog's nats InForce).
type busAnswer struct {
	Allowed      bool   `json:"allowed"`
	InForce      *bool  `json:"in_force"`
	Why          string `json:"why"`
	FileWritten  string `json:"file_written"`
	ServerLoaded string `json:"server_loaded"`
}

func readBusAnswer(raw json.RawMessage) (busAnswer, error) {
	var answer busAnswer
	err := json.Unmarshal(raw, &answer)
	return answer, err
}

// grantLookEvery is how often the bus's user list is judged against the one its machine was last sent.
const grantLookEvery = 30 * time.Second

// channelsIn is what the channels are now, as a fingerprint: each module claiming a kind of the channel
// bench, where, promising what, and whether of its own account. An ask the router refused is asked again
// once this changes.
func channelsIn(inv *inventory.Inventory) func(ctx context.Context) string {
	return func(ctx context.Context) string {
		entries, err := inv.Catalogued(ctx)
		if err != nil {
			return ""
		}
		var parts []string
		for _, e := range entries {
			for _, c := range e.Manifest.Claims {
				if c.Kind == "" || !catalogue.KindedBenches[c.Name] {
					continue
				}
				on := append([]string(nil), e.On...)
				sort.Strings(on)
				caps := append([]string(nil), c.Capabilities...)
				sort.Strings(caps)
				parts = append(parts, fmt.Sprintf("%s/%s=%s@%s[%s]own:%t", c.Name, c.Kind, e.Manifest.Module,
					strings.Join(on, ","), strings.Join(caps, ","), e.Manifest.RunsAs != ""))
			}
		}
		sort.Strings(parts)
		return strings.Join(parts, ";")
	}
}

// startAsking makes the serving controller's asker and hands it the router's words.
func startAsking(ctx context.Context, open *stores, server *link.Server, conn *nats.Conn, keeper *conditions.Keeper) {
	js, err := jetstream.New(conn)
	if err != nil {
		fmt.Printf("the operator cannot be asked: %v\n", err)
		return
	}
	a := servingAsker(open, conn, js, keeper)
	if err := server.Decides(a); err != nil {
		fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
		return
	}
	askerFrom = a
	go a.keep(ctx)
}

// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
// records, and the judgements it asks by — the bus's grant to ask among them (novox/hq issue 353).
func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper *conditions.Keeper) *asker {
	return &asker{
		open: keeper.Open,
		silence: func(ctx context.Context, key string, d time.Duration, by, why string) error {
			_, err := keeper.Silence(ctx, key, d, by, why)
			return err
		},
		store: busAsked{conn: conn},
		publish: func(ctx context.Context, subject string, body []byte, id string) error {
			_, err := js.Publish(ctx, subject, body, jetstream.WithMsgID(id))
			return err
		},
		call: callAction(conn),
		// A proposed settings layer is set by this controller itself on the warrant (novox/hq ADR 0277).
		setLayer: setLayerIn(open),
		record: func(ctx context.Context, act link.HandAct) error {
			_, err := link.RecordHandAct(ctx, conn, act)
			return err
		},
		routerRecord: routerRecordOf(conn, open.inventory),
		routerHere:   routerHereIn(open.inventory),
		grantHeld:    grantHeldIn(open, conn),
		channels:     channelsIn(open.inventory),
		raise: func(ctx context.Context, obs []conditions.Observation) error {
			return keeper.Reconcile(ctx, sourceAsker, obs)
		},
		now:  time.Now,
		logf: func(format string, args ...any) { fmt.Printf(format+"\n", args...) },
	}
}