// Package groups is what the forge bridge's process and its tools bundle share of a delivery group's end (novox/hq ADR // 0633 §4, issue 1585): the keys of the bridge's state performed, and the asks of the state end-asks, through which the // tool bridge_group_end_again asks the process to say a group's end again (the bridge serves no request subject, ADR 0437 // §4, so a tool asks in a state the process watches, as indexer_take_in asks the indexer in takeins). package groups import ( "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "slices" "sort" "strconv" "strings" "time" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" ) // The bridge's states the tools read and write. const ( // Module is the forge bridge's own name (its manifest's module, checked by the bridge's names test). Module = "mesh-delivery-forge-bridge" // PerformedState is the bridge's state of what it performed and published. PerformedState = "performed" // EndAsksState is the state of asks to say a group's end again, keyed by the gate's merges key of the group. EndAsksState = "end-asks" // EndAsksBucket is its bucket on the bus. EndAsksBucket = Module + "_" + EndAsksState ) // Key is the key of performed that says a group's end is written: `groups.`, its value the revision of the // gate's merges key it was worked out from. func Key(mergesKey string) string { return "groups." + mergesKey } // PublishedKey is the key of performed that says a member's merged message is written: `published.<16 hex digits of the // sha256 of #@>`. func PublishedKey(repository string, number int, merge string) string { sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d@%s", strings.ToLower(repository), number, merge))) return "published." + hex.EncodeToString(sum[:8]) } // EndID is the message id of a group's end, and of one said again by hand at a time: the stream's duplicate window drops a // second publish of the same id. func EndID(group, lastMerge, again string) string { id := "group-end." + group + "." + lastMerge if again != "" { id += "." + again } return id } // AgainStamp is the time a hand ask appends to a group end's message id. func AgainStamp(at time.Time) string { return at.UTC().Format("20060102T150405.000000000Z") } // EndAsk is a value of the end-asks state: which group's end a person asked to be said again, why, who and when, and, once // the bridge read it, its answer. type EndAsk struct { V int `json:"v"` Group string `json:"group"` Why string `json:"why,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` Written bool `json:"written,omitempty"` Answer string `json:"answer,omitempty"` Answered time.Time `json:"answered,omitzero"` } // Done says the ask is answered. func (a EndAsk) Done() bool { return !a.Answered.IsZero() } // NewEndAsk is the value bridge_group_end_again writes for a group: its key (the gate's merges key) and the value. func NewEndAsk(group, why, by string, at time.Time) (string, []byte, error) { group = strings.TrimSpace(group) if group == "" { return "", nil, errors.New("name the group whose end is said again (group)") } key, err := meshdeliveryclient.MergesKey(group) if err != nil { return "", nil, err } data, err := json.Marshal(EndAsk{V: 1, Group: group, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) return key, data, err } // DecodeEndAsk reads an end-asks value. func DecodeEndAsk(data []byte) (EndAsk, error) { var a EndAsk if err := json.Unmarshal(data, &a); err != nil { return a, err } if a.V != 1 || a.Group == "" { return a, errors.New("an end-asks value of no version 1 naming no group") } return a, nil } // Row is one group as bridge_groups answers it: the gate's merge of it and what the bridge kept of its end. type Row struct { Key string `json:"key"` Group string `json:"group"` Members []RowMember `json:"members"` Ended bool `json:"ended"` EndedAt uint64 `json:"ended-at-revision,omitempty"` Open bool `json:"open"` Words string `json:"words"` // LastEndAsk is the last ask to say the group's end again (bridge_group_end_again), its who, why and when, and the // bridge's answer once it read it; absent when none was asked. The state end-asks keeps the last per group. LastEndAsk *EndAsk `json:"last-end-ask,omitempty"` } // RowMember is one approved member of a group: its merge's state at the gate and whether its merged message is written. type RowMember struct { Repository string `json:"repository"` Number int `json:"number"` State string `json:"state"` Commit string `json:"commit,omitempty"` Published bool `json:"published"` } // Rows are the groups the gate's merges hold with more than one member, each with what the bridge kept: ended (its end // written, the revision it was worked out from) or open (some member is not merged or its merged message not written, or its // end not written). merges is the gate's merges by key, performed the bridge's performed keys by key; unread are the gate's // keys that do not read. asks are the end-asks values by the same key; one that does not read is named unread as // `end-asks.`. func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[string]uint64, asks map[string][]byte) (rows []Row, unread []string) { lastAsk := map[string]*EndAsk{} for key, raw := range asks { a, err := DecodeEndAsk(raw) if err != nil { unread = append(unread, EndAsksState+"."+key) continue } lastAsk[key] = &a } for key, raw := range merges { m, err := meshdeliveryclient.DecodeMerge(raw) if err != nil { unread = append(unread, key) continue } if len(m.Approved) < 2 { continue } r := Row{Key: key, Group: m.Group} allMerged := true for _, p := range m.Pulls { pub := p.Commit != "" && performed[PublishedKey(p.Repository, int(p.Number), p.Commit)] r.Members = append(r.Members, RowMember{Repository: p.Repository, Number: int(p.Number), State: p.State, Commit: p.Commit, Published: pub}) if meshdeliveryclient.MergeState(p.State) != meshdeliveryclient.MergeMerged || !pub { allMerged = false } } if len(m.Pulls) < len(m.Approved) { allMerged = false } r.EndedAt, r.Ended = endedAt[key] r.LastEndAsk = lastAsk[key] r.Open = !r.Ended switch { case r.Ended: r.Words = "its end is written" case allMerged: r.Words = "every member is merged and its message written, and its end is not written: bridge_group_end_again says it" default: r.Words = "gathered: not every member is merged and its merged message written" } rows = append(rows, r) } sortRows(rows) sort.Strings(unread) return rows, unread } func sortRows(rows []Row) { sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key }) } // MembersOf are a merge's approved members with their heads, in the order of their keys. func MembersOf(m seat.MergesByGroupValue) []meshdeliveryclient.Head { var out []meshdeliveryclient.Head for _, k := range slices.Sorted(func(yield func(string) bool) { for k := range m.Approved { if !yield(k) { return } } }) { at := strings.LastIndex(k, "@") if at < 0 { continue } n, err := strconv.Atoi(k[at+1:]) if err != nil { continue } out = append(out, meshdeliveryclient.Head{Repository: k[:at], Number: n, Head: m.Approved[k]}) } return out }