mesh-delivery/sdk/sourcebranches/sourcebranches.go

// Package sourcebranches is the follower of the branches the seat mesh-delivery's verb source keeps (design 47 G §614),
// in the Delivery domain's SDK so that a seat's client holds no other seat's types (ADR 0625 §4): the first view of the
// verb given no branch, then each source-moved heard, and the asking again until a change heard is held.
package sourcebranches

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"time"

	delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/keys"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// Seat is the seat whose verb source and event source-moved are followed.
const Seat = delivery.Seat

// EventSourceMoved is the event the seat says when a branch's source moved.
const EventSourceMoved = delivery.EventSourceMoved

// SourceVerb is the gate's verb a part reads the branches the gate keeps a source revision of by, before the first
// registrations, as its manifest's invokes names it (design 47 G §614).
const SourceVerb = "seat:" + delivery.Seat + "." + delivery.VerbSource

// StateSourceBranches names the view of the branches the gate's source keeps in a part's reader (`mesh-delivery.source`).
const StateSourceBranches = "source"

// Call asks one seat verb over the bus and answers what the holder said, whatever wraps it (process.Conn.Call, or
// stdio.Ask through the tool runner).
type Call func(key string, body any) (json.RawMessage, error)

// Created stands for the picture's Created in a follower: followed by the seat's event, never by an owner's views family,
// so one fixed time, never zero.
var Created = time.Unix(0, 0).UTC()

// ClipBytes is data cut to 200 bytes for a message.
func ClipBytes(b []byte) string {
	if len(b) > 200 {
		return string(b[:200]) + "…"
	}
	return string(b)
}

// SourceBranches is a part's follower of the branches the gate's `source` keeps (design 47 G §614): a first view of the
// verb `source` given no branch, then each `source-moved` heard. Its picture is keyed by the escaped branch name.
type SourceBranches struct {
	Follower *views.Follower

	call Call
}

// New is a SourceBranches asking the gate through call. Nothing is asked before the reader starts it.
func New(call Call) *SourceBranches {
	s := &SourceBranches{call: call}
	s.Follower = &views.Follower{Owner: Seat, Tool: SourceVerb, State: StateSourceBranches, Ask: s.ask}
	return s
}

// Branches is the escaped name of each branch the picture holds.
func (s *SourceBranches) Branches() []string {
	out := make([]string, 0)
	for b := range s.Follower.Values() {
		out = append(out, b)
	}
	slices.Sort(out)
	return out
}

// Hear takes one `source-moved` event: a branch the picture does not hold asks the first view again until it does, on
// its own goroutine, so the consumer is never held. Data that is no such event is an error.
func (s *SourceBranches) Hear(ctx context.Context, data []byte) error {
	var ev delivery.SourceMovedEvent
	if err := json.Unmarshal(data, &ev); err != nil || ev.Branch == "" {
		return fmt.Errorf("sourcebranches: not a %s event: %s", EventSourceMoved, ClipBytes(data))
	}
	held := func() bool { _, ok := s.Follower.Get(keys.EscapeName(ev.Branch)); return ok }
	if held() {
		return nil
	}
	go CatchUp(ctx, s.Follower, held, fmt.Sprintf("source of %s moved", ev.Branch))
	return nil
}

func (s *SourceBranches) ask(_ context.Context, _ string) (views.Answer, error) {
	raw, err := s.call(SourceVerb, delivery.SourceInput{})
	if err != nil {
		return views.Answer{}, err
	}
	answer, said, ok := buscalls.Unwrap(raw)
	if !ok {
		return views.Answer{}, fmt.Errorf("%s answered: %s", SourceVerb, said)
	}
	var a delivery.SourceAnswer
	if err := json.Unmarshal(answer, &a); err != nil {
		return views.Answer{}, fmt.Errorf("%s answered no branches: %s", SourceVerb, ClipBytes(answer))
	}
	if a.Error != "" {
		return views.Answer{}, fmt.Errorf("%s answered: %s", SourceVerb, a.Error)
	}
	values := make(map[string]json.RawMessage, len(a.Branches))
	for _, b := range a.Branches {
		values[b] = json.RawMessage(`{}`)
	}
	return views.Answer{V: 1, Owner: Seat, State: StateSourceBranches, Created: Created, Values: values, At: time.Now()}, nil
}

// CatchUp asks f's first view again until held says it holds what was heard, pausing between asks as the follower's Wait
// says (doubling to a minute), or until ctx ends: an ask already running may have been answered before the change was
// written.
func CatchUp(ctx context.Context, f *views.Follower, held func() bool, why string) {
	wait := f.Wait
	if wait == 0 {
		wait = 5 * time.Second
	}
	for !held() {
		if err := f.Gap(ctx, why); err != nil {
			return
		}
		if held() {
			return
		}
		p, cancel := context.WithTimeout(ctx, wait)
		<-p.Done()
		cancel()
		if ctx.Err() != nil {
			return
		}
		if wait *= 2; wait > time.Minute {
			wait = time.Minute
		}
	}
}