mesh-controller/internal/inventory/bindings.go

package inventory

import (
	"context"
	"time"

	"github.com/novox/mesh-controller/internal/catalogue"
)

// A binding to data (novox/hq ADR 0232): where each consumer of a provision that keeps its data was
// last sent, so a resolution that would answer it from anywhere else keeps it there instead.

// Binding is one consumer's recorded provider for one provision.
type Binding struct {
	// Machine is where the consumer runs, Consumer the module there.
	Machine   string
	Consumer  string
	Provision string
	Provider  catalogue.Chosen
	BoundAt   time.Time
	SentAt    time.Time
	// MovedFrom is where it was bound before a pin moved it, empty when never moved.
	MovedFrom string
}

// BindingsFor is every binding recorded for a machine's consumers, by consumer then provision — the
// shape the resolver reads (catalogue.World.Bound).
func (i *Inventory) BindingsFor(ctx context.Context, machine string) (map[string]map[string]catalogue.Chosen, error) {
	node, err := i.NodeByName(ctx, machine)
	if err != nil {
		return nil, err
	}
	rows, err := i.store.Pool().Query(ctx,
		`select consumer, provision, provider_node, provider_module from binding where node = $1`, node.ID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]map[string]catalogue.Chosen{}
	for rows.Next() {
		var consumer, provision string
		var c catalogue.Chosen
		if err := rows.Scan(&consumer, &provision, &c.Node, &c.Module); err != nil {
			return nil, err
		}
		if out[consumer] == nil {
			out[consumer] = map[string]catalogue.Chosen{}
		}
		out[consumer][provision] = c
	}
	return out, rows.Err()
}

// Bindings is every binding recorded, by machine, consumer and provision.
func (i *Inventory) Bindings(ctx context.Context) ([]Binding, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select n.name, b.consumer, b.provision, b.provider_node, b.provider_module, b.bound_at, b.sent_at,
		        coalesce(b.moved_from, '')
		   from binding b join node n on n.id = b.node
		  order by n.name, b.consumer, b.provision`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Binding
	for rows.Next() {
		var b Binding
		if err := rows.Scan(&b.Machine, &b.Consumer, &b.Provision, &b.Provider.Node, &b.Provider.Module,
			&b.BoundAt, &b.SentAt, &b.MovedFrom); err != nil {
			return nil, err
		}
		out = append(out, b)
	}
	return out, rows.Err()
}

// RecordBindings writes down the bindings a declaration just sent to a machine carried. A binding
// already recorded at the same provider is only marked sent; one recorded elsewhere — moved by a pin,
// the only way the resolver lets one move — keeps where it was in moved_from and is bound anew.
func (i *Inventory) RecordBindings(ctx context.Context, machine string, bindings []Binding) error {
	if len(bindings) == 0 {
		return nil
	}
	node, err := i.NodeByName(ctx, machine)
	if err != nil {
		return err
	}
	tx, err := i.store.Pool().Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	for _, b := range bindings {
		_, err := tx.Exec(ctx,
			`insert into binding (node, consumer, provision, provider_node, provider_module)
			 values ($1, $2, $3, $4, $5)
			 on conflict (node, consumer, provision) do update set
			   sent_at = now(),
			   moved_from = case
			     when binding.provider_node = excluded.provider_node
			      and binding.provider_module = excluded.provider_module then binding.moved_from
			     else binding.provider_node || '/' || binding.provider_module end,
			   bound_at = case
			     when binding.provider_node = excluded.provider_node
			      and binding.provider_module = excluded.provider_module then binding.bound_at
			     else now() end,
			   provider_node = excluded.provider_node,
			   provider_module = excluded.provider_module`,
			node.ID, b.Consumer, b.Provision, b.Provider.Node, b.Provider.Module)
		if err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}