mesh-catalog/modules/nats/cmd/nats-tools/keep.go

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/handacts"
	"git.novox.be/novox/mesh-sdk/go/process"

	"nats-tools/internal/bussnap"
	"nats-tools/internal/keeper"
)

// The keeper (`nats-tools keep`): the bus module's process on its own account (own secret broker-keeper, ADR 0436 §1),
// started by the node-engine as the manifest's process keeper. It asks the indexer's first view once, then hears
// manifest-accepted on its durable consumer, makes what each asks for, reads the fill every minute, and acts on the
// persons' requests its tools put in its own state requests (requests.go). It serves nothing (ADR 0436 §3).

// AcceptedOf reads a manifest-accepted body: mesh-sdk's delivery.ManifestAccepted, with the indexer's revision of the
// verdict when the event carries it (a field mesh-sdk is asked to add; 0 until then).
func AcceptedOf(body []byte) (keeper.Accepted, error) {
	var a struct {
		delivery.ManifestAccepted
		Revision uint64 `json:"revision"`
	}
	if err := json.Unmarshal(body, &a); err != nil {
		return keeper.Accepted{}, err
	}
	if a.Module == "" {
		return keeper.Accepted{}, fmt.Errorf("a manifest-accepted names no module")
	}
	return keeper.Accepted{Module: a.Module, Digest: a.Digest, Revision: a.Revision, BusState: a.BusState}, nil
}

// AcceptedSubject is the indexer's manifest-accepted on the shared events stream.
var AcceptedSubject = "mesh.mod." + keeper.FirstViewOwner + ".event." + delivery.ManifestAcceptedEvent

// The router's records of asks, read under this module's own name (ADR 0259 §3), as the mail sender reads them.
const asksBucket = "messenger_asks"

// processStore is the keeper's own state through its process connection.
type processStore struct{ c *process.Conn }

func (s processStore) Get(state, key string) ([]byte, uint64, bool, error) {
	e, err := s.c.State(state).Get(key)
	if err != nil || e == nil {
		return nil, 0, false, err
	}
	return e.Value, e.Revision, true, nil
}

func (s processStore) Put(state, key string, value []byte) (uint64, error) {
	return s.c.State(state).Put(key, json.RawMessage(value))
}

func (s processStore) Delete(state, key string) error { return s.c.State(state).Delete(key) }

func (s processStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() }

// streamEmitter says an event on the bus module's own stream, with the SDK's headers.
func streamEmitter(js jetstream.JetStream, a process.Account) func(string, keeper.Event) error {
	return func(event string, body keeper.Event) error {
		data, err := json.Marshal(body)
		if err != nil {
			return err
		}
		msg := nats.NewMsg(a.StreamSubject(event))
		id := eventID()
		msg.Header.Set("x-event-id", id)
		msg.Header.Set("x-source", a.Module)
		msg.Header.Set("x-node", a.Node)
		msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
		msg.Header.Set("content-type", "application/json")
		msg.Data = data
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
		return err
	}
}

func eventID() string {
	var b [16]byte
	_, _ = rand.Read(b[:])
	return hex.EncodeToString(b[:])
}

// sayer raises and clears the keeper's conditions through the condition keeper, as every module says them.
func sayer(js jetstream.JetStream) func(asks.Condition) error {
	return func(c asks.Condition) error {
		subject, err := asks.ConditionSubject(keeper.Module)
		if err != nil {
			return err
		}
		data, err := asks.EncodeCondition(c)
		if err != nil {
			return err
		}
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_, err = js.Publish(ctx, subject, data)
		return err
	}
}

// firstView asks the indexer's indexer_bus_state as the keeper's own account.
func firstView(c *process.Conn) func() (keeper.FirstView, error) {
	return func() (keeper.FirstView, error) {
		raw, err := c.Call(keeper.FirstViewOwner+"."+keeper.FirstViewVerb, map[string]any{})
		if err != nil {
			return keeper.FirstView{}, err
		}
		return FirstViewOf(raw)
	}
}

// FirstViewOf reads indexer_bus_state's answer, as the tool runner hands it back ({result, error} or bare).
func FirstViewOf(raw []byte) (keeper.FirstView, error) {
	var wrapped struct {
		Result json.RawMessage `json:"result"`
		Error  string          `json:"error"`
	}
	if json.Unmarshal(raw, &wrapped) == nil && (wrapped.Result != nil || wrapped.Error != "") {
		if wrapped.Error != "" {
			return keeper.FirstView{}, fmt.Errorf("%s", wrapped.Error)
		}
		raw = wrapped.Result
	}
	var v keeper.FirstView
	if err := json.Unmarshal(raw, &v); err != nil {
		return keeper.FirstView{}, fmt.Errorf("the answer is not a first view: %w", err)
	}
	return v, nil
}

func percent(getenv func(string) string, name string, fallback int64) int64 {
	if n, err := strconv.ParseInt(strings.TrimSpace(getenv(name)), 10, 64); err == nil && n > 0 && n < 100 {
		return n
	}
	return fallback
}

// runKeeper is the keeper's process: it returns when ctx ends, or when it has no credential.
func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "nats-keeper: ", log.LstdFlags)
	account, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return err
	}
	conn, err := process.Connect(account)
	if err != nil {
		return err
	}
	defer conn.Close()
	nc, js, err := account.Dial(nats.Name(account.Node + "." + account.Module + ".keeper"))
	if err != nil {
		return err
	}
	defer nc.Close()
	recorder := handacts.Recorder{Ask: conn.Call}
	k := &keeper.Keeper{
		Bus:   keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
		Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
		Act: func(what, why, cause string) error {
			_, err := recorder.Record(handacts.Request{What: what, Why: why, Cause: cause})
			return err
		},
		Warrant: func(id string) (*asks.Ask, *asks.Warrant, error) {
			e, err := conn.Ask(asksBucket, id)
			if err != nil || e == nil {
				return nil, nil, err
			}
			var rec struct {
				Ask     *asks.Ask     `json:"ask"`
				Warrant *asks.Warrant `json:"warrant"`
			}
			if err := json.Unmarshal(e.Value, &rec); err != nil {
				return nil, nil, err
			}
			return rec.Ask, rec.Warrant, nil
		},
		Logf:  logger.Printf,
		Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70),
	}
	logger.Printf("started on the account %s: asking %s.%s for the first view", account.User, keeper.FirstViewOwner, keeper.FirstViewVerb)
	return Keep(ctx, k, hearer(js, account, logger.Printf), requestsOf(conn), logger.Printf)
}

// hearer binds the keeper's durable consumer on EVENTS (<node>_nats_keeper, filtered on manifest-accepted) and hands
// each message to handle, which acknowledges it once applied.
func hearer(js jetstream.JetStream, a process.Account, logf func(string, ...any)) func(context.Context, func(subject string, body []byte) error) error {
	return func(ctx context.Context, handle func(string, []byte) error) error {
		cons, err := js.Consumer(ctx, process.EventsStream, a.QueueConsumerName())
		if err != nil {
			return fmt.Errorf("the keeper's consumer %s on %s: %w", a.QueueConsumerName(), process.EventsStream, err)
		}
		cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
			if err := handle(m.Subject(), m.Data()); err != nil {
				logf("%v: delivered again", err)
				_ = m.NakWithDelay(30 * time.Second)
				return
			}
			_ = m.Ack()
		}, logf)
		if err != nil {
			return err
		}
		go func() { <-ctx.Done(); cc.Stop() }()
		return nil
	}
}

// Keep is the keeper's loop, apart from the bus so a test drives it: the first view asked until it answers, on its
// backoff; then hear, which applies each manifest-accepted; a fill pass every minute; the waiting entries made again
// on the same backoff; and the persons' requests acted on as they come. It returns when ctx ends.
func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func(string, []byte) error) error,
	requests func(context.Context, func()) error, logf func(string, ...any)) error {
	var wait time.Duration
	for {
		next, err := k.AskFirstView(wait)
		if err == nil {
			break
		}
		logf("the first view did not answer (%v): nothing is created or orphaned until it does; asked again in %s", err, next)
		wait = next
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(next):
		}
	}
	logf("the first view answered at revision %d", k.Status().FirstView.Revision)
	heard := make(chan struct{}, 1)
	poke := func() {
		select {
		case heard <- struct{}{}:
		default:
		}
	}
	for backoff := 10 * time.Second; ; backoff = min(backoff*2, 5*time.Minute) {
		err := hear(ctx, func(subject string, body []byte) error {
			if subject != AcceptedSubject && !strings.HasSuffix(subject, ".event."+delivery.ManifestAcceptedEvent) {
				return nil
			}
			a, err := AcceptedOf(body)
			if err != nil {
				logf("a manifest-accepted cannot be read (%v): taken, and nothing made of it", err)
				return nil
			}
			k.Heard(a)
			poke()
			return nil
		})
		if err == nil {
			break
		}
		logf("manifest-accepted cannot be heard yet (%v): asked again in %s", err, backoff)
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(backoff):
		}
	}
	asked := make(chan struct{}, 1)
	if requests != nil {
		if err := requests(ctx, func() {
			select {
			case asked <- struct{}{}:
			default:
			}
		}); err != nil {
			logf("the persons' requests cannot be watched (%v): acted on at each fill pass", err)
		}
	}
	tick := time.NewTicker(time.Minute)
	defer tick.Stop()
	retry := 10 * time.Second
	var again <-chan time.Time
	k.FillPass()
	for {
		if len(k.Waiting()) > 0 && again == nil {
			again = time.After(retry)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-tick.C:
			k.FillPass()
			// A recreate or a drop waits for the operator's warrant: read again once a minute, as an asker reads its asks.
			k.ActOnRequests()
		case <-asked:
			k.ActOnRequests()
		case <-heard:
		case <-again:
			again = nil
			k.CheckAll()
			if len(k.Waiting()) > 0 {
				retry = min(retry*2, 5*time.Minute)
				logf("%d entr(y/ies) wait to be made: %s; made again in %s", len(k.Waiting()), strings.Join(k.Waiting(), "; "), retry)
			} else {
				retry = 10 * time.Second
			}
		}
	}
}

// keepMain runs the keeper until a signal.
func keepMain() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := runKeeper(ctx, os.Getenv, os.Stderr); err != nil && ctx.Err() == nil {
		fmt.Fprintln(os.Stderr, "nats-keeper:", err)
		os.Exit(1)
	}
}

// requestsOf watches the keeper's own state requests, waking act at each change.
func requestsOf(c *process.Conn) func(context.Context, func()) error {
	return func(_ context.Context, act func()) error {
		return c.State(keeper.StateRequests).Watch("", func(stdio.StateChange) error {
			act()
			return nil
		})
	}
}