mesh-tools/node-tools/internal/runtime/runtime.go

// Package runtime is the node's tool runtime (novox/hq ADR 0175, ADR 0193), the Go port of
// node-tools' runtime.ts in its launch-only form: it launches every assigned module's bundle,
// serves each module's tools on that module's subjects and each held seat's verbs on the seat's,
// and answers for every module the verb that says what it serves. It imports nothing and knows no
// language.
package runtime

import (
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"

	"github.com/novox/mesh-tools/node-tools/internal/announce"
	"github.com/novox/mesh-tools/node-tools/internal/bus"
	"github.com/novox/mesh-tools/node-tools/internal/launch"
)

// ToolsVerb is the verb every module's runtime answers for it (ADR 0152): its tools, from the code
// that answers them.
const ToolsVerb = "tools"

// Words the mesh sets for the runtime (ADR 0175, ADR 0192).
const (
	ToolModules     = "MESH_TOOL_MODULES"
	ToolEnv         = "MESH_TOOL_ENV"
	OperatorAccount = "MESH_OPERATOR_ACCOUNT"
	OperatorHome    = "MESH_OPERATOR_HOME"
)

// Served is one module this runtime serves and its entrypoints.
type Served struct {
	Module      string
	Entrypoints []string
}

// ToolsAnswer is what `tools` answers for one module.
type ToolsAnswer struct {
	Module string       `json:"module"`
	Tools  []ListedTool `json:"tools"`
	Failed string       `json:"failed,omitempty"`
}

// ListedTool is one tool as a module's `tools` answer lists it.
type ListedTool struct {
	Name        string          `json:"name"`
	Description string          `json:"description"`
	Input       json.RawMessage `json:"input"`
	Subjects    []string        `json:"subjects,omitempty"`
}

// ServedModulesFrom reads MESH_TOOL_MODULES: `<module>=<entrypoint>` entries, comma-separated, several
// per module. The one-module form — a bare path, or the runtime's own module — is the per-module
// containers' (to-be 38 WP4c) and refused here: the node's runtime imports nothing.
func ServedModulesFrom(spec, own string) ([]Served, error) {
	order := []string{}
	by := map[string][]string{}
	for _, raw := range strings.Split(spec, ",") {
		entry := strings.TrimSpace(raw)
		if entry == "" {
			continue
		}
		module, path, ok := strings.Cut(entry, "=")
		module, path = strings.TrimSpace(module), strings.TrimSpace(path)
		if !ok || module == "" || path == "" || module == own {
			return nil, fmt.Errorf("%s: %q is not <module>=<entrypoint> of another module; the node's "+
				"runtime launches the bundles it is given and imports nothing (novox/hq ADR 0193)", ToolModules, entry)
		}
		if _, seen := by[module]; !seen {
			order = append(order, module)
		}
		by[module] = append(by[module], path)
	}
	out := make([]Served, 0, len(order))
	for _, m := range order {
		out = append(out, Served{Module: m, Entrypoints: by[m]})
	}
	return out, nil
}

// TakeToolEnvs reads the composed environments (ADR 0192) and removes them from the process's, so no
// bundle finds another's there.
func TakeToolEnvs() (map[string]map[string]string, error) {
	raw := os.Getenv(ToolEnv)
	os.Unsetenv(ToolEnv)
	out := map[string]map[string]string{}
	if raw == "" {
		return out, nil
	}
	var parsed map[string]map[string]any
	if err := json.Unmarshal([]byte(raw), &parsed); err != nil {
		return nil, fmt.Errorf(`%s is not JSON of the shape {"<module>": {"<word>": "<value>"}}: %w`, ToolEnv, err)
	}
	for module, words := range parsed {
		own := map[string]string{}
		for k, v := range words {
			if s, ok := v.(string); ok {
				own[k] = s
			} else {
				b, _ := json.Marshal(v)
				own[k] = string(b)
			}
		}
		out[module] = own
	}
	return out, nil
}

type registration struct {
	module string // the module's own name, or a seat's
	owner  string // the module whose bundle made it
	tools  []launch.Tool
}

// Run launches, binds and serves. It answers a stop function.
func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, logf func(string, ...any)) (func(), error) {
	if account := os.Getenv(OperatorAccount); account != "" {
		home := ""
		if h := os.Getenv(OperatorHome); h != "" {
			home = " (home " + h + ")"
		}
		logf("[mesh-tools] the operator's account here is %s%s", account, home)
	}
	modules := make([]string, 0, len(served))
	isServed := map[string]bool{}
	for _, s := range served {
		modules = append(modules, s.Module)
		isServed[s.Module] = true
		conn.Follow(s.Module)
	}
	node := conn.Node()

	base := os.Environ()
	envFor := func(module string) []string {
		words := map[string]string{}
		for _, kv := range base {
			if k, v, ok := strings.Cut(kv, "="); ok && k != ToolEnv {
				words[k] = v
			}
		}
		for k, v := range envs[module] {
			words[k] = v
		}
		words["MESH_SERVED_MODULE"] = module
		words["MESH_MODULE"] = module
		if node != "" {
			words["MESH_NODE"] = node
		}
		out := make([]string, 0, len(words))
		for k, v := range words {
			out = append(out, k+"="+v)
		}
		sort.Strings(out)
		return out
	}

	failed := map[string]string{}
	var registrations []registration
	var stops []func()
	for _, s := range served {
		for _, entry := range s.Entrypoints {
			path, _ := filepath.Abs(entry)
			module := s.Module
			fail := func(why string) {
				failed[module] = why
				logf("[mesh-tools] %s's bundle %s failed to load: %s; its tools are not served here", module, entry, why)
			}
			if !launch.Executable(path) {
				fail(path + " is not executable; a bundle the runtime serves is started, never imported, and its build makes it executable (novox/hq ADR 0193)")
				continue
			}
			child, err := launch.Start(module, path, envFor(module), func(params json.RawMessage) error {
				var env bus.Envelope
				if err := json.Unmarshal(params, &env); err != nil {
					return fmt.Errorf("not an event envelope: %w", err)
				}
				return conn.PublishAs(module, env)
			}, logf)
			if err != nil {
				fail(err.Error())
				continue
			}
			stops = append(stops, child.Stop)
			for _, r := range child.Registrations {
				registrations = append(registrations, registration{module: r.Module, owner: module, tools: r.Tools})
			}
		}
	}

	claimed := map[string]bool{}
	for _, m := range modules {
		if mem := conn.Membership(m); mem != nil {
			for _, s := range mem.Seats {
				claimed[s.Seat] = true
			}
		}
	}
	var own []registration
	for _, r := range registrations {
		switch {
		case isServed[r.module]:
			own = append(own, r)
		case claimed[r.module]:
		default:
			logf(`[mesh-tools] %s registers tools under "%s", which is neither a module served here nor a seat one of them claims; not served until the mesh issues the claim`, r.owner, r.module)
		}
	}
	stopAll := func() {
		for i := len(stops) - 1; i >= 0; i-- {
			stops[i]()
		}
	}
	for _, r := range own {
		seen := map[string]bool{}
		for _, t := range r.tools {
			if t.Name == ToolsVerb {
				stopAll()
				return nil, fmt.Errorf(`%s names a tool "%s", which is the verb the runtime answers for every module with what it serves (novox/hq ADR 0152) — refused, rename it`, r.module, ToolsVerb)
			}
			if seen[t.Name] {
				stopAll()
				return nil, fmt.Errorf("%s exposes two tools named %s — refused", r.module, t.Name)
			}
			seen[t.Name] = true
		}
	}

	var names []string
	byModule := map[string][]launch.Tool{}
	for _, r := range own {
		for _, t := range r.tools {
			t := t
			stop, err := conn.Handle(r.module+"."+t.Name, func(body json.RawMessage) (any, error) {
				return t.Run(argsOf(body))
			})
			if err != nil {
				logf("[mesh-tools] cannot serve %s.%s: %v", r.module, t.Name, err)
				continue
			}
			names = append(names, r.module+"."+t.Name)
			stops = append(stops, stop)
		}
		byModule[r.module] = append(byModule[r.module], r.tools...)
	}
	for _, module := range modules {
		module := module
		tools := byModule[module]
		why := failed[module]
		if len(tools) == 0 && why == "" {
			continue // a pure-events module: silent, as it always was
		}
		stop, err := conn.Handle(module+"."+ToolsVerb, func(json.RawMessage) (any, error) {
			answer := ToolsAnswer{Module: module, Tools: []ListedTool{}, Failed: why}
			for _, t := range tools {
				answer.Tools = append(answer.Tools, ListedTool{Name: t.Name, Description: t.Description,
					Input: t.Input, Subjects: subjectsOf(conn, module, t.Name)})
			}
			return answer, nil
		})
		if err == nil {
			stops = append(stops, stop)
		}
	}

	failedNames := make([]string, 0, len(failed))
	for _, m := range modules {
		if _, f := failed[m]; f {
			failedNames = append(failedNames, m)
		}
	}
	line := fmt.Sprintf("[mesh-tools] serving %d tool(s) for %d module(s): %s", len(names), len(modules), orNone(names))
	if len(failedNames) > 0 {
		line += fmt.Sprintf("; not serving %s, whose bundle(s) failed to load", strings.Join(failedNames, ", "))
	}
	logf("%s", line)
	stops = append(stops, serveSeats(conn, modules, registrations, logf))

	// **What it serves, it announces** (novox/hq ADR 0197): the NATS services protocol's discovery,
	// answered with what is served at the moment it is asked — re-served memberships included.
	announced, err := announce.Serve(conn, announce.Service{
		Name: conn.Module(), ID: instanceOf(conn),
		Description: "the mesh's tool runtime on " + node + ": every assigned module's tools and the seats they hold",
		Metadata:    map[string]string{"node": node},
	}, func() []announce.Endpoint { return endpointsOf(conn, own, registrations, modules) })
	if err != nil {
		logf("[mesh-tools] cannot announce what it serves: %v", err)
	} else {
		stops = append(stops, announced)
	}
	conn.Flush()
	return stopAll, nil
}

// instanceOf is this runtime's instance on the bus: its machine, which is what tells two instances of
// one service apart; the connection's module where it has no machine.
func instanceOf(conn *bus.Conn) string {
	if n := conn.Node(); n != "" {
		return n
	}
	return conn.Module()
}

// endpointsOf is everything this runtime serves now: each served module's tools on every subject the
// mesh issued for them, and each held seat's verbs on the seat's subject (ADR 0197).
func endpointsOf(conn *bus.Conn, own []registration, registrations []registration, modules []string) []announce.Endpoint {
	node := conn.Node()
	var out []announce.Endpoint
	for _, r := range own {
		for _, t := range r.tools {
			served := conn.ServedOn(r.module, t.Name)
			interchangeable := false
			for _, s := range served {
				interchangeable = interchangeable || s.Subject == "mesh.mod."+r.module+".tool."+t.Name
			}
			for _, s := range served {
				out = append(out, announce.Endpoint{Kind: announce.KindTool, Module: r.module, Tool: t.Name,
					Node: node, Description: t.Description, Schema: t.Input, Interchangeable: interchangeable,
					Subject: s.Subject, Queue: s.Queue})
			}
		}
	}
	impl := map[string]map[string]launch.Tool{}
	for _, r := range registrations {
		if impl[r.module] == nil {
			impl[r.module] = map[string]launch.Tool{}
		}
		for _, t := range r.tools {
			impl[r.module][t.Name] = t
		}
	}
	have := map[string]bool{}
	for _, module := range modules {
		m := conn.Membership(module)
		if m == nil {
			continue
		}
		for _, v := range m.Seats {
			t, ok := impl[v.Seat][v.Verb]
			if !ok || have[v.Subject] {
				continue
			}
			have[v.Subject] = true
			scope := "mesh"
			if node != "" && strings.HasSuffix(v.Subject, "."+node) {
				scope = "node"
			}
			out = append(out, announce.Endpoint{Kind: announce.KindSeat, Module: module, Tool: v.Verb, Seat: v.Seat,
				Scope: scope, Node: node, Description: t.Description, Schema: t.Input, Subject: v.Subject})
		}
	}
	return out
}

func orNone(names []string) string {
	if len(names) == 0 {
		return "(none)"
	}
	return strings.Join(names, ", ")
}

// argsOf is a call's arguments as the tool receives them: an object, `{}` for none.
func argsOf(body json.RawMessage) json.RawMessage {
	trimmed := strings.TrimSpace(string(body))
	if trimmed == "" || trimmed == "null" {
		return json.RawMessage("{}")
	}
	return body
}

// subjectsOf is where a tool is answered as the mesh issued it: the plain subject first, then this
// machine's; nothing before a membership is issued.
func subjectsOf(conn *bus.Conn, module, tool string) []string {
	m := conn.Membership(module)
	if m == nil {
		return nil
	}
	var plain, mine []string
	for _, s := range m.Serves {
		subject := strings.ReplaceAll(s.Subject, "{tool}", tool)
		if s.Queue != "" {
			plain = append(plain, subject)
		} else {
			mine = append(mine, subject)
		}
	}
	return append(plain, mine...)
}

// serveSeats serves every verb of every seat a served module holds, where the mesh issued it, by
// the tool of the same name registered under the seat's name (ADR 0159, 0160) — and serves again
// whenever a membership changes. Whether this machine holds the seat is the bus's to decide.
func serveSeats(conn *bus.Conn, modules []string, registrations []registration, logf func(string, ...any)) func() {
	impl := map[string]map[string]launch.Tool{}
	for _, r := range registrations {
		if impl[r.module] == nil {
			impl[r.module] = map[string]launch.Tool{}
		}
		for _, t := range r.tools {
			impl[r.module][t.Name] = t
		}
	}
	var mu sync.Mutex
	var stops []func()
	serve := func() {
		mu.Lock()
		defer mu.Unlock()
		for _, s := range stops {
			s()
		}
		stops = nil
		have := map[string]bool{}
		for _, module := range modules {
			m := conn.Membership(module)
			if m == nil {
				continue
			}
			for _, v := range m.Seats {
				if have[v.Subject] {
					continue
				}
				have[v.Subject] = true
				t, ok := impl[v.Seat][v.Verb]
				if !ok {
					logf("[mesh-tools] %s claims %s and implements no %s, which that seat promises; not served", module, v.Seat, v.Verb)
					continue
				}
				stop, err := conn.HandleSubject(v.Subject, func(body json.RawMessage) (any, error) {
					return t.Run(argsOf(body))
				})
				if err != nil {
					logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)
					continue
				}
				stops = append(stops, stop)
				logf("[mesh-tools] serving %s's %s on %s, admitted where %s holds the seat", v.Seat, v.Verb, v.Subject, module)
			}
		}
	}
	serve()
	conn.OnMembership(func(bus.Membership) { go func() { serve(); conn.Flush() }() })
	return func() {
		mu.Lock()
		defer mu.Unlock()
		for _, s := range stops {
			s()
		}
		stops = nil
	}
}