mesh-host/cmd/mesh-host/filling.go

package main

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"sync/atomic"

	"git.novox.be/novox/mesh-sdk/go/declaration"
	"git.novox.be/novox/mesh-sdk/go/provisions"
	"github.com/novox/mesh-host/internal/apply"
	"github.com/novox/mesh-host/internal/filling"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/store"
	"github.com/novox/mesh-host/internal/system"
)

// serveFilling writes filled files on the provisioner client's request for as long as this engine serves (novox/hq
// ADR 0349 §4). A socket that cannot be served is said once, and this engine goes on applying.
func serveFilling(ctx context.Context, opts options, signer ed25519.PublicKey, queue *link.Session, node string, say func(string)) {
	e := fillingEngine(opts, signer, provisionSecretsOn(queue, node, opts.state), say)
	fillingNow.Store(e)
	defer fillingNow.CompareAndSwap(e, nil)
	if err := filling.Serve(ctx, filling.Socket, e, say); err != nil {
		say("filled files are not written on request on this machine: " + err.Error())
	}
}

// fillingNow is the filling this engine serves, for knockFilling; nil while none is served.
var fillingNow atomic.Pointer[filling.Engine]

// knockFilling tells the provisioner's client to fill again what the filling refused or failed and the declaration
// applied now may let through (filling.Engine.Knock), after what runs changed: a module's apply or removal settled
// (agent.Env.Settled, "<module> settled"), the haves were read at the start, or a module was taken as running
// (agent.Env.RunsChanged, issue 924). What it told, or why it could not, is said.
func knockFilling(after string, say func(string)) {
	e := fillingNow.Load()
	if e == nil {
		return
	}
	told, err := e.Knock(filling.Socket)
	switch {
	case err != nil:
		say(fmt.Sprintf("after %s: %v", after, err))
	case told != "":
		say(fmt.Sprintf("after %s, the provisioner's client is told to fill again: %s", after, told))
	}
}

// fillingEngine is the write of a filled file as this engine does it: the declaration it keeps, verified, read under
// the apply's lock — the same two locks every apply takes, the process's and the one beside the state — and the file
// written through the apply (apply.WriteFilled), its outcome recorded in the node's state as an apply's is.
//
// A provision's `${secret:}` the client left in the content is filled first (secrets, novox/hq ADR 0570 clause 6), and a
// file so filled is written 0600 when the declaration gives no mode.
func fillingEngine(opts options, signer ed25519.PublicKey, secrets provisionSecrets, say func(string)) *filling.Engine {
	return &filling.Engine{
		Declared: func() (*declaration.Declaration, error) {
			return declaredNow(opts.state, signer)
		},
		UIDOf: filling.UIDOf,
		Lock: func() (func(), error) {
			applying.Lock()
			unlock, err := store.Lock(opts.state, nil)
			if err != nil {
				applying.Unlock()
				return nil, err
			}
			return func() { unlock(); applying.Unlock() }, nil
		},
		Place: func(ctx context.Context, d *declaration.Declaration, path string, content []byte) (string, string, error) {
			built, err := system.For(builtFor)
			if err != nil {
				return "", "", err
			}
			known, err := store.Load(opts.state)
			if err != nil {
				return "", "", err
			}
			content, secret, err := secrets.fill(ctx, filledModule(d, path), content)
			if err != nil {
				return "", "", fmt.Errorf("%s was not written: %w", path, err)
			}
			report, updated, writeErr := apply.WriteFilledWith(ctx, built, d, known, path, content, apply.ExecRunner,
				announceOr(say), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state)),
				apply.Filled{Secret: secret})
			// Saved whichever way it went, as an apply's: what was written is on the machine either way.
			if err := store.Save(opts.state, updated); err != nil {
				return "", "", errors.Join(writeErr, errors.New("written, and the node's state could not be saved: "+
					err.Error()))
			}
			action, digest := "", ""
			for _, o := range report.Outcomes {
				if o.Target == path && o.Type == string(declaration.TypeFile) {
					action = o.Action
				}
			}
			if rec, ok := updated.At(string(declaration.TypeFile), path); ok {
				digest = rec.Wrote
			}
			return action, digest, writeErr
		},
	}
}

// filledModule is the module whose file the declaration declares at path, or "" when none or several do.
func filledModule(d *declaration.Declaration, path string) string {
	module := ""
	for _, r := range d.Resources {
		if f, ok := r.(*declaration.File); ok && f.Path == path {
			if module != "" {
				return ""
			}
			module, _ = d.ModuleOfID(f.ID)
		}
	}
	return module
}

// provisionSecretsOn reads provisions and asks the vault over the link, and opens with this machine's sealing key.
func provisionSecretsOn(queue *link.Session, node, statePath string) provisionSecrets {
	p := provisionSecrets{node: node, open: sealOpener(statePath)}
	if queue == nil {
		return p
	}
	// What the bus holds under prefix, read by a watch to its caught-up mark: no poll, no list of every key.
	p.list = func(ctx context.Context, prefix string) (map[string][]byte, error) {
		wctx, cancel := context.WithCancel(ctx)
		defer cancel()
		updates, err := queue.KVWatch(wctx, provisions.ProvisionsBucket, prefix+">")
		if err != nil {
			return nil, err
		}
		out := map[string][]byte{}
		for {
			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case u, ok := <-updates:
				if !ok {
					return nil, fmt.Errorf("the watch of %s ended before it caught up", provisions.ProvisionsBucket)
				}
				switch {
				case u.Caught:
					return out, nil
				case u.Deleted:
					delete(out, u.Key)
				default:
					out[u.Key] = u.Value
				}
			}
		}
	}
	p.own = func(ctx context.Context, args map[string]any) (json.RawMessage, error) {
		return queue.AskVault(ctx, "own", args)
	}
	return p
}