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..chunk...`, each acknowledged by the caller on // `mesh.seat..ack...`. 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 }