mesh-host/internal/selfcompose/packages.go

package selfcompose

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// **The packages this node's compositions declare, kept by its engine** (novox/hq ADR 0565 §6, §15; issue 1953). The
// composer keeps the package resources of each composition it handed, and after a pass that handed something new (or
// the run's first) it works out the node's key of the engine's state packages from them (delivery.DeclaredPackagesOf,
// in the layout and order the controller's mirror had) and writes it only when the value differs from what the key holds (ADR 0289
// §1): no timer, and no composing of another node. A module held keeps the packages of what was handed of it, as it keeps
// running what it runs. pacman reads the key as mesh-host.packages; the node seat's verb packages answers it, and with
// write writes it again (ADR 0565 §15).

// PackagesStore is this node's key of the engine's state packages (link.EnginePackages).
type PackagesStore interface {
	// Get is the key's value, nil when it has none.
	Get(ctx context.Context, key string) ([]byte, error)
	Put(ctx context.Context, key string, value []byte) error
}

// packagesOf is a composition's package resources alone, with its module: what the node's key is worked out from.
func packagesOf(comp delivery.Composed) delivery.Composed {
	out := delivery.Composed{Node: comp.Node, Module: comp.Module}
	for _, raw := range comp.Resources {
		var head struct {
			Type string `json:"type"`
		}
		if json.Unmarshal(raw, &head) == nil && head.Type == "package" {
			out.Resources = append(out.Resources, raw)
		}
	}
	return out
}

// DeclaredPackages is the node's key as this composer would write it now, from what it handed.
func (c *Composer) DeclaredPackages() (delivery.DeclaredPackages, error) {
	c.mu.Lock()
	keys := make([]string, 0, len(c.pkgs))
	for k := range c.pkgs {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	comps := make([]delivery.Composed, 0, len(keys))
	for _, k := range keys {
		comps = append(comps, c.pkgs[k])
	}
	composed := c.pkgsComposed
	c.mu.Unlock()
	if !composed {
		return delivery.DeclaredPackages{}, errors.New("this node's composer has handed no composition yet in this run, so it " +
			"declares no packages yet")
	}
	return delivery.DeclaredPackagesOf(c.env.Node, comps)
}

// publishPackages writes the node's key of the state packages when a pass asked it (pkgsDue) and the value differs.
func (c *Composer) publishPackages(ctx context.Context) {
	c.mu.Lock()
	due := c.pkgsDue
	c.pkgsDue = false
	c.mu.Unlock()
	if !due || c.env.Packages == nil {
		return
	}
	wrote, err := c.writePackages(ctx, false)
	if err != nil {
		c.sayPackages("this node's packages are not kept in " + enginePackagesBucket + "; " + c.owePackages() + ": " + err.Error())
		return
	}
	c.mu.Lock()
	c.pkgsTries = 0
	c.mu.Unlock()
	c.sayPackages("")
	if wrote {
		c.env.Say("this node's packages: its key of " + enginePackagesBucket + " written")
	}
}

// enginePackagesBucket is the engine's state packages as the bus names it, for what the composer says.
const enginePackagesBucket = "mesh-host_" + delivery.EnginePackagesState

// WritePackages writes the node's key again from what the composer handed, when it differs from what the bus holds
// now (the node seat's verb packages with write, ADR 0565 §15): the key is read again rather than trusted, so a key lost
// or changed on the bus is written back. It answers whether it wrote.
func (c *Composer) WritePackages(ctx context.Context) (bool, error) {
	if c.env.Packages == nil {
		return false, errors.New("this engine keeps no packages key")
	}
	return c.writePackages(ctx, true)
}

// writePackages writes the key when the value worked out differs from what it holds: read once per run (pkgsKept), or
// again when fresh says so.
func (c *Composer) writePackages(ctx context.Context, fresh bool) (bool, error) {
	v, err := c.DeclaredPackages()
	if err != nil {
		return false, err
	}
	raw, err := json.Marshal(v)
	if err != nil {
		return false, err
	}
	c.pkgsWrite.Lock()
	defer c.pkgsWrite.Unlock()
	if fresh || !c.pkgsRead {
		kept, err := c.env.Packages.Get(ctx, c.env.Node)
		if err != nil {
			return false, fmt.Errorf("%s cannot be read: %w", c.env.Node, err)
		}
		c.pkgsKept, c.pkgsRead = kept, true
	}
	if bytes.Equal(c.pkgsKept, raw) {
		return false, nil
	}
	if err := c.env.Packages.Put(ctx, c.env.Node, raw); err != nil {
		return false, err
	}
	c.pkgsKept = raw
	return true, nil
}

// owePackages keeps the packages due and has the loop make a pass again after the growing bound, as oweData does for the
// data items: the first write after a delivery may be refused while the state or its grant is not on the bus yet.
func (c *Composer) owePackages() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.pkgsDue = true
	c.wait = LaterWait(c.pkgsTries)
	c.pkgsTries++
	c.owed.Store(true)
	return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.pkgsTries)
}

// sayPackages says a line about the packages once until it changes; "" clears it.
func (c *Composer) sayPackages(line string) {
	c.mu.Lock()
	same := c.pkgsSaid == line
	c.pkgsSaid = line
	c.mu.Unlock()
	if !same && line != "" {
		c.env.Say(line)
	}
}