Files
mesh-delivery/modules/mesh-grants/internal/buscompose/chunks.go
T
2026-11-12 00:00:23 +01:00

96 lines
3.5 KiB
Go

package buscompose
import (
"slices"
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// **The chunk subjects of a verb that sends a file** (design 47 G §681, A-45, A-60): a verb whose seat definition says `chunks: true`
// has its answer's file sent in chunks on `mesh.seat.<seat>.chunk.<verb>.<call>.<n>`, each acknowledged by the caller on
// `mesh.seat.<seat>.ack.<verb>.<call>.<n>`. They are granted to the process that serves the verb (publish the chunks, subscribe to
// the acknowledgements) and, reversed, to an account that names the verb by name in `invokes`; a `*` entry grants none.
// ChunkVerb is one verb of a seat whose definition says it sends chunks.
type ChunkVerb struct{ Seat, Verb string }
// ChunkVerbsOf is every verb the seat definitions say send chunks, sorted. Pure.
func ChunkVerbsOf(defs map[string]seatdef.Definition) []ChunkVerb {
var out []ChunkVerb
for _, d := range defs {
for _, v := range d.Verbs {
if v.Chunks {
out = append(out, ChunkVerb{Seat: d.Seat, Verb: v.Name})
}
}
}
sort.Slice(out, func(i, j int) bool {
if out[i].Seat != out[j].Seat {
return out[i].Seat < out[j].Seat
}
return out[i].Verb < out[j].Verb
})
return out
}
func chunkSubject(c ChunkVerb) string { return "mesh.seat." + c.Seat + ".chunk." + c.Verb + ".>" }
func ackSubject(c ChunkVerb) string { return "mesh.seat." + c.Seat + ".ack." + c.Verb + ".>" }
// ackServed is what the process that answers a chunked verb subscribes: the acknowledgements of every call and chunk (A-109).
func ackServed(c ChunkVerb) string { return "mesh.seat." + c.Seat + ".ack." + c.Verb + ".*.*" }
// chunkGrants are the chunk subjects p is granted: as a process that serves a chunked verb (`serves`, design 47 G A-109), the
// publish of its chunks and the subscription to their acknowledgements, and as an account that names a chunked verb by name in
// `invokes`, the subscription to the chunks and the publish of the acknowledgements. The chunks of a snapshot are a transfer on
// the core bus: no stream, and any caller of the verb hears them.
func chunkGrants(p Principal) (pub, sub []string) {
if len(p.Chunked) == 0 {
return nil, nil
}
for _, c := range p.Chunked {
if p.Kind == KindProcess && slices.Contains(p.AnswersSeat, "seat:"+c.Seat+"."+c.Verb) {
pub, sub = append(pub, chunkSubject(c)), append(sub, ackServed(c))
}
if slices.Contains(p.Invokes, "seat:"+c.Seat+"."+c.Verb) {
sub, pub = append(sub, chunkSubject(c)), append(pub, ackSubject(c))
}
}
return pub, sub
}
// restoreGrants are the keeper's standing grant of the bus's stream restore subjects and their chunk subjects, for the bus names
// of the streams ADR 0436 §3 restricts and only those (design 47 G §681, A-60): the restore request of each and the chunks the
// snapshot is sent to.
func restoreGrants(names []string) []string {
var out []string
for _, n := range names {
if !safeSubject.MatchString(n) || strings.Contains(n, ".") {
continue
}
out = append(out, "$JS.API.STREAM.RESTORE."+n, "$JS.SNAPSHOT.RESTORE."+n+".>")
}
return out
}
// restrictedStreamNames are the bus names of the streams every placed module declares restricted (ADR 0363 §4), sorted, once each.
func restrictedStreamNames(r Records) []string {
seen := map[string]bool{}
for _, mods := range r.Assigned {
for _, d := range mods {
for _, s := range d.Streams {
if s.Restricted {
seen[s.Name] = true
}
}
}
}
out := make([]string, 0, len(seen))
for n := range seen {
out = append(out, n)
}
sort.Strings(out)
return out
}