mesh-controller/internal/broker/seattraffic.go

package broker

import "sort"

// What a module may say and take on the seats it holds, uses and watches, beyond tools (novox/hq ADR
// 0259 §3, to-be 46 §10).
//
// Three rules are added to the ones a seat always had, each a subject whose last token names who may
// publish it, granted to that publisher alone — the way a node seat's event about a machine carries the
// machine (ADR 0219):
//
//   - **A verb named by its caller** (`by-caller`). A user of the seat submits that accept, and hears that
//     event, under its own module's name and no other: `accept.ask.<module>`, `event.decided.<module>`. The
//     holder takes every caller's accept and says the event to any caller. So an ask's asker is a fact the
//     server enforces, and a warrant reaches only the asker it is for.
//   - **A kinded bench** (ADR 0234 §2). A holder claims one kind and takes its own kind's accepts, says its
//     own kind's events and asks its own kind's proofs, and nothing of another kind; a user submits to any
//     kind. Each kind has its own worker on the seat's queue, so a holder that is away keeps its work and
//     holds up no other kind.
//   - **A proof** (`proofs`): core request and reply on `mesh.seat.<seat>.proof.<verb>.<kind>`. No stream's
//     subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
//     holder asks with its own kind; the modules that watch the seat answer.
//
// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's.

// KindedBenches are the seats that may be kinded (ADR 0234 §2): making another is a decision, recorded.
var KindedBenches = map[string]bool{"channel": true, "intake": true}

// WorkerName is the worker a seat's holders pull from: one for the seat, or one per kind on a kinded bench.
func WorkerName(seat, kind string) string {
	if kind == "" {
		return "SEAT_" + upperSnake(seat) + "_worker"
	}
	return "SEAT_" + upperSnake(seat) + "_" + upperSnake(kind) + "_worker"
}

func namesVerb(list []string, s string) bool {
	for _, x := range list {
		if x == s {
			return true
		}
	}
	return false
}

// SeatTrafficOf derives one module's seat traffic from the seats it holds, uses and watches. Only seats
// carrying one of the rules above are read: every other seat is composed as it always was.
func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
	var t SeatTraffic
	for _, s := range holds {
		if !s.isNewTraffic() {
			continue
		}
		kind := ""
		if s.Kinded {
			kind = s.Kind
			if kind == "" || !safeSubject.MatchString(kind) {
				// A kinded claim without a usable kind is refused at registration; here it is granted
				// nothing, which is the same answer at the last place it could be asked.
				continue
			}
		}
		if len(s.Accepts) > 0 {
			stream := seatStreamName(s.Name)
			filter := "mesh.seat." + s.Name + ".accept.>"
			if kind != "" {
				filter = "mesh.seat." + s.Name + ".accept.*." + kind
			}
			t.Workers = append(t.Workers, Worker{Stream: stream, Consumer: WorkerName(s.Name, kind), Filter: filter})
		}
		for _, a := range s.Accepts {
			switch {
			case kind != "":
				t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+"."+kind))
			case namesVerb(s.ByCaller, a):
				t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+".*"))
			}
		}
		for _, e := range s.Emits {
			switch {
			case kind != "":
				t.Publish = append(t.Publish, seatSubject(s, "event", e+"."+kind))
			case namesVerb(s.ByCaller, e):
				t.Publish = append(t.Publish, seatSubject(s, "event", e+".*"))
			}
		}
		if kind != "" {
			for _, v := range s.Proofs {
				t.Publish = append(t.Publish, seatSubject(s, "proof", v+"."+kind))
			}
		}
	}
	for _, s := range uses {
		if !s.isNewTraffic() {
			continue
		}
		for _, a := range s.Accepts {
			switch {
			case namesVerb(s.ByCaller, a):
				t.Publish = append(t.Publish, seatSubject(s, "accept", a+"."+module))
			case s.Kinded && module == s.DeclaredBy:
				// Work for a kind is put on its queue by the bench's own router, and by no other user.
				t.Publish = append(t.Publish, seatSubject(s, "accept", a+".*"))
			}
		}
		for _, e := range s.Emits {
			if namesVerb(s.ByCaller, e) {
				t.Subscribe = append(t.Subscribe, seatSubject(s, "event", e+"."+module))
			}
		}
		for _, b := range s.Records {
			if !safeSubject.MatchString(b) {
				continue
			}
			t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
		}
	}
	for _, w := range watches {
		if w.Kinded {
			for _, e := range w.Emits {
				t.Subscribe = append(t.Subscribe, seatSubject(w, "event", e+".*"))
			}
			if module == w.DeclaredBy {
				// A code is answered by the bench's own router, and by no other watcher.
				for _, v := range w.Proofs {
					t.Answers = append(t.Answers, seatSubject(w, "proof", v+".*"))
				}
			}
		}
	}
	t.Publish = unique(t.Publish)
	t.Subscribe = unique(t.Subscribe)
	t.Answers = unique(t.Answers)
	t.Records = unique(t.Records)
	sort.Slice(t.Workers, func(i, j int) bool { return t.Workers[i].Consumer < t.Workers[j].Consumer })
	return t
}

// seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
// worker is pulled and acknowledged through and a record is read through.
func seatTrafficGrants(t SeatTraffic) (pub, sub []string) {
	pub = append(pub, t.Publish...)
	sub = append(sub, t.Subscribe...)
	sub = append(sub, t.Answers...)
	for _, w := range t.Workers {
		pub = append(pub,
			"$JS.API.CONSUMER.INFO."+w.Stream+"."+w.Consumer,
			"$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Consumer,
			"$JS.ACK."+w.Stream+"."+w.Consumer+".>")
	}
	pub = append(pub, t.Records...)
	return pub, sub
}

// isNewTraffic says whether a seat carries any of the rules above, so a seat that carries none is
// composed exactly as before them.
func (s Seat) isNewTraffic() bool {
	return s.Kinded || len(s.ByCaller) > 0 || len(s.Proofs) > 0 || len(s.Records) > 0
}

// plainVerbs is a seat's accepts or emits with those the rules above compose taken out: a verb named by
// its caller and every verb of a kinded bench are composed by SeatTrafficOf and nowhere else.
func plainVerbs(s Seat, verbs []string) []string {
	if s.Kinded {
		return nil
	}
	var out []string
	for _, v := range verbs {
		if !namesVerb(s.ByCaller, v) {
			out = append(out, v)
		}
	}
	return out
}

// SeatTrafficObjects is the work queues and workers the seat traffic of every composed user implies
// (novox/hq ADR 0259 §3): a queue for each seat a holder takes work from, and each holder's worker on it —
// one per kind on a kinded bench, filtered to that kind, so the kinds never take each other's work. Only
// seats carrying the rules above; the mesh's own seats' queues are RaiseSeats'.
func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) {
	streams := map[string]Stream{}
	consumers := map[string]Consumer{}
	add := func(module string, holds []Seat) {
		for _, w := range SeatTrafficOf(module, holds, nil, nil).Workers {
			seat := ""
			for _, s := range holds {
				if seatStreamName(s.Name) == w.Stream {
					seat = s.Name
				}
			}
			streams[w.Stream] = Stream{
				Name:      w.Stream,
				Subjects:  []string{"mesh.seat." + seat + ".accept.>"},
				Retention: RetentionWorkQueue,
				MaxAge:    7 * 24 * 60 * 60,
				Why:       "work submitted to the " + seat + " seat; its holders take it, each kind its own, and it queues while nobody does",
			}
			consumers[w.Consumer] = Consumer{
				Name:           w.Consumer,
				Stream:         w.Stream,
				Filters:        []string{w.Filter},
				AckWaitSeconds: 60,
				// No bound on redelivery: a channel away for a day keeps its work, offered again later and
				// later by its holder's runtime (novox/hq ADR 0259; the correctness review of 2026-10-08).
				MaxDeliver: 0,
				Why: module + " holds " + seat + "; it pulls one ask at a time and acknowledges once it has " +
					"recorded it, so a crash redelivers rather than loses",
			}
		}
	}
	for _, p := range users {
		switch p.Kind {
		case KindModule:
			add(p.Module, p.Holds)
		case KindNodeTools:
			for _, d := range p.Carries {
				add(d.Module, d.Holds)
			}
		}
	}
	var ss []Stream
	for _, s := range streams {
		ss = append(ss, s)
	}
	sort.Slice(ss, func(i, j int) bool { return ss[i].Name < ss[j].Name })
	var cs []Consumer
	for _, c := range consumers {
		cs = append(cs, c)
	}
	sort.Slice(cs, func(i, j int) bool { return cs[i].Name < cs[j].Name })
	return ss, cs
}

// trustedTraffic is why a module's seat traffic is the trusted holder's (novox/hq ADR 0259 §8), or "": it
// says a seat's event to one caller each (a warrant), or holds a kind of a kinded bench that proves its sender.
func trustedTraffic(d Declared) string {
	for _, s := range d.Holds {
		for _, e := range s.Emits {
			if namesVerb(s.ByCaller, e) {
				return "it says " + s.Name + "'s " + e + " to one caller each"
			}
		}
		if s.Kinded && namesVerb(s.Capabilities, "verified-sender") {
			return "it holds " + s.Name + " of kind " + s.Kind + ", which proves its sender"
		}
	}
	return ""
}

// TrafficQueues is the work queue of every seat naming its caller or its kind that accepts work, held or not
// (novox/hq ADR 0259 §3): what is submitted before a holder is assigned waits for it.
func TrafficQueues(seats []Seat) []Stream {
	var out []Stream
	seen := map[string]bool{}
	for _, s := range seats {
		if len(s.Accepts) == 0 || !s.isNewTraffic() || seen[s.Name] {
			continue
		}
		seen[s.Name] = true
		out = append(out, Stream{Name: seatStreamName(s.Name), Subjects: []string{"mesh.seat." + s.Name + ".accept.>"},
			Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60,
			Why: "work submitted to the " + s.Name + " seat; its holders take it, each kind its own, and it queues while nobody does"})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}