mesh-delivery/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go

// Command mesh-delivery-teller-tools serves the teller's recovery tool (issue 774): the runtime launches it for a call.
// teller_end_ask writes one value of the teller's state ends and waits for the teller's answer there. The teller is the
// other program: it serves no request subject (ADR 0437 §4).
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"

	"mesh-delivery-teller/internal/ends"
)

// asker writes an ends value and waits for the teller's answer to it.
type asker func(key string, value []byte) (ends.End, error)

// Tools are the teller's tools.
func Tools(ask asker, now func() time.Time) []stdio.Tool {
	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
	return []stdio.Tool{
		{Name: ends.Tool,
			Description: "End one open ask the delivery teller holds at the operator channel, by hand, saying why: the teller " +
				"cancels it at the router, which frees one of the three places the router gives the teller, or drops it from the " +
				"teller's wait for room. For an ask whose question is gone but whose key was never taken away (a merge ask of a pull " +
				"request merged by another path, a backup ask about a backup no longer failing), which otherwise holds its place " +
				"until it lapses, so a newer ask is refused for want of room. Name the ask by its router id (as " +
				"operator-channel's asks show it) or by the key of the gate's or the assigner's asks. Answers which ask it found, " +
				"whether it ended it, and what it did; an ask answered, lapsed or ended already is left and said so.",
			Input: map[string]any{"type": "object", "required": []string{"ask", "why"}, "properties": map[string]any{
				"ask": map[string]any{"type": "string", "description": "the ask's router id, or the key of the gate's or the assigner's asks it was asked for"},
				"why": map[string]any{"type": "string", "description": "why it is ended by hand, in plain words"}}},
			Run: func(args map[string]any) (any, error) {
				key, value, err := ends.New(str(args, "ask"), str(args, "why"), stdio.CallerOf(args), now())
				if err != nil {
					return nil, err
				}
				e, err := ask(key, value)
				if err != nil {
					return nil, err
				}
				return e.Result, nil
			}},
	}
}

// askEnd writes the ends value at key and waits, on a watch of the key, at most bound for the teller's answer.
func askEnd(s stdio.KeptState, bound time.Duration) asker {
	return func(key string, value []byte) (ends.End, error) {
		answered := make(chan ends.End, 1)
		var rev uint64
		var mu sync.Mutex
		take := func(revision uint64, v []byte) {
			mu.Lock()
			mine := rev
			mu.Unlock()
			if mine == 0 || revision <= mine {
				return
			}
			e, err := ends.Decode(v)
			if err != nil || !e.Done() {
				return
			}
			select {
			case answered <- e:
			default:
			}
		}
		if err := s.Watch(key, func(c stdio.StateChange) error {
			if c.Op == "put" {
				take(c.Revision, c.Value)
			}
			return nil
		}); err != nil {
			return ends.End{}, fmt.Errorf("watching %s.%s: %w", ends.State, key, err)
		}
		r, err := s.Put(key, json.RawMessage(value))
		if err != nil {
			return ends.End{}, fmt.Errorf("the end could not be written to %s: %w", ends.Bucket, err)
		}
		mu.Lock()
		rev = r
		mu.Unlock()
		// The watch may have heard the answer before the revision was known: read it once too.
		if e, err := s.Get(key); err == nil && e != nil {
			take(e.Revision, e.Value)
		}
		wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8)
		defer cancel()
		select {
		case e := <-answered:
			return e, nil
		case <-wait.Done():
			return ends.End{}, fmt.Errorf("the teller did not answer within %s: is it running? its answer will be in %s.%s", bound, ends.Bucket, key)
		}
	}
}

func main() {
	if err := stdio.Serve("", Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now)); err != nil {
		fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err)
		os.Exit(1)
	}
}