mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/sent.go

package main

// Every operation's event is sent once, after a restart too (novox/hq to-be 55 §1): the event of a change carries the
// version it follows as its message id, so the bus drops a second send; the last version sent per record is a key of the
// state `sent`; and at start the events of every later operation are sent, so a module stopped between an operation and its
// event loses none, and a consumer of `reviewed` never waits for one that was never sent.

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

const sentState = "sent"

func sentKey(k Key) string { return fmt.Sprintf("%s.%04d", k.Kind, k.Number) }

// markSent keeps that the events up to version were sent. It only moves forward.
func (k *KB) markSent(key Key, version int) {
	if k.States == nil {
		return
	}
	kv := k.States.State(sentState)
	for try := 0; try < 3; try++ {
		e, err := kv.Get(sentKey(key))
		if err != nil {
			k.logf("[mesh-knowledge] what was sent of %s cannot be read: %v", key, err)
			return
		}
		if e == nil {
			if _, err = kv.Create(sentKey(key), version); err == nil {
				return
			}
			continue
		}
		var have int
		_ = json.Unmarshal(e.Value, &have)
		if have >= version {
			return
		}
		if _, err = kv.Update(sentKey(key), version, e.Revision); err == nil {
			return
		}
	}
}

// sentUpTo is the last version of a record whose events were sent.
func (k *KB) sentUpTo(key Key) int {
	if k.States == nil {
		return 0
	}
	e, err := k.States.State(sentState).Get(sentKey(key))
	if err != nil || e == nil {
		return 0
	}
	var have int
	_ = json.Unmarshal(e.Value, &have)
	return have
}

// OperationInfo is one operation of a record's history, as far as sending its event needs.
type OperationInfo struct {
	Sequence  uint64
	Operation string
	Version   int
	By        string
	At        time.Time
	Why       string
}

// OperationReader is a store that keeps every change as an operation: the ledger.
type OperationReader interface {
	// Operations are the operations of a record that made a version, oldest first.
	Operations(ctx context.Context, k Key) ([]OperationInfo, error)
}

// eventOf is the event an operation says; none for an operation whose change is not an event of its own.
var eventOf = map[string]string{OpReserve: EventReserved, OpPropose: EventProposed, OpImport: EventImported, OpReview: EventReviewed,
	OpAccept: EventAccepted, OpSubmit: EventSubmitted, OpTakeOver: EventTakenOver}

// ResendEvents sends, at start, the event of every operation after the last one sent of each record, and answers how
// many it sent. The same message id as the first send, so a send that did arrive is dropped by the bus.
func (k *KB) ResendEvents(ctx context.Context) (int, error) {
	r, ok := k.Store.(OperationReader)
	if !ok || k.EmitID == nil {
		return 0, nil
	}
	all, err := k.Store.All(ctx)
	if err != nil {
		return 0, err
	}
	sent := 0
	for _, e := range all {
		key := e.Current.Key()
		from := k.sentUpTo(key)
		if from >= e.Versions {
			continue
		}
		ops, err := r.Operations(ctx, key)
		if err != nil {
			return sent, err
		}
		for _, op := range ops {
			if op.Version <= from {
				continue
			}
			if event := eventOf[op.Operation]; event != "" {
				body := map[string]any{"kind": key.Kind, "number": key.Number, "record": key.String(), "by": op.By, "at": op.At,
					"what": op.Why, "version": op.Version, "status": e.Current.Status}
				if err := k.send(event, key, op.Version, body); err != nil {
					return sent, fmt.Errorf("the event %s for %s could not be sent: %v", event, key, err)
				}
				sent++
			}
		}
		k.markSent(key, e.Versions)
	}
	return sent, nil
}