bldr/plugin/dir-load-plugin.go

Rama master · réplica de lectura del repositorio original.

package bldr_plugin

import (
	"context"
	"slices"

	"github.com/aperturerobotics/controllerbus/bus"
	"github.com/aperturerobotics/controllerbus/directive"
	"github.com/aperturerobotics/starpc/srpc"
	"github.com/aperturerobotics/util/routine"
	"github.com/pkg/errors"
	manifest "github.com/s4wave/spacewave/bldr/manifest"
	"github.com/s4wave/spacewave/net/hash"
)

// LoadPlugin is a directive to execute a plugin by ID using the best available host.
type LoadPlugin interface {
	// Directive indicates LoadPlugin is a directive.
	directive.Directive

	// LoadPluginID returns the plugin ID to load.
	// Cannot be empty.
	LoadPluginID() string

	// LoadPluginInstanceKey returns the instance key for instanced plugins.
	// When non-empty, (plugin_id, instance_key) is the composite deduplication
	// key, creating a separate worker per instance.
	// When empty, a single shared instance per plugin_id is used.
	LoadPluginInstanceKey() string

	// LoadPluginManifestRoot selects an exact manifest content hash.
	// Empty follows the latest eligible manifest. A missing exact root never falls back.
	LoadPluginManifestRoot() string

	// LoadPluginManifests selects the installation and its recovery artifacts, newest first.
	// Replacement uses prepared admission; nil follows normal catalog selection.
	LoadPluginManifests() []*manifest.ManifestRef

	// LoadPluginServeAssets reports that the caller serves the plugin's files
	// to a browser. When no host can run the plugin, the scheduler mounts the
	// files of its browser manifest without running it.
	LoadPluginServeAssets() bool
}

// LoadPluginValue is the result type for LoadPlugin.
// Multiple results may be pushed to the directive.
type LoadPluginValue = RunningPlugin

// loadPlugin implements LoadPlugin
type loadPlugin struct {
	pluginID     string
	instanceKey  string
	manifestRoot string
	manifests    []*manifest.ManifestRef
	serveAssets  bool
}

// NewLoadPlugin constructs a new LoadPlugin directive.
func NewLoadPlugin(pluginID string) LoadPlugin {
	return &loadPlugin{pluginID: pluginID}
}

// NewLoadPluginInstanced constructs a new LoadPlugin directive with an instance key.
func NewLoadPluginInstanced(pluginID, instanceKey string) LoadPlugin {
	return &loadPlugin{pluginID: pluginID, instanceKey: instanceKey}
}

// NewLoadPluginAtManifest requests an immutable executable within an existing binding.
func NewLoadPluginAtManifest(pluginID, instanceKey, manifestRoot string) LoadPlugin {
	return &loadPlugin{pluginID: pluginID, instanceKey: instanceKey, manifestRoot: manifestRoot}
}

// NewLoadPluginWithManifests selects an installation and its recovery artifacts.
func NewLoadPluginWithManifests(pluginID, instanceKey string, refs ...*manifest.ManifestRef) LoadPlugin {
	cloned := make([]*manifest.ManifestRef, len(refs))
	for i, ref := range refs {
		cloned[i] = ref.CloneVT()
	}
	return &loadPlugin{pluginID: pluginID, instanceKey: instanceKey, manifests: cloned}
}

// NewLoadPluginAssets loads a plugin whose files the caller serves to a browser.
func NewLoadPluginAssets(pluginID string) LoadPlugin {
	return &loadPlugin{pluginID: pluginID, serveAssets: true}
}

// NewLoadPluginAssetsAtManifest mounts an exact browser manifest without running it.
// The directive publishes its manifest root once the files are available.
func NewLoadPluginAssetsAtManifest(pluginID, manifestRoot string) LoadPlugin {
	return &loadPlugin{pluginID: pluginID, manifestRoot: manifestRoot, serveAssets: true}
}

// ExLoadPlugin executes the LoadPlugin directive.
//
// if returnIfIdle=true and the directive becomes idle, returns nil, nil, nil
func ExLoadPlugin(
	ctx context.Context,
	b bus.Bus,
	returnIfIdle bool,
	pluginID string,
	valDisposeCallback func(),
) (LoadPluginValue, directive.Instance, directive.Reference, error) {
	return bus.ExecWaitValue[RunningPlugin](
		ctx,
		b,
		NewLoadPlugin(pluginID),
		bus.ReturnIfIdle(returnIfIdle),
		valDisposeCallback,
		nil,
	)
}

// ExLoadPluginInstanced executes the LoadPlugin directive with an instance key.
func ExLoadPluginInstanced(
	ctx context.Context,
	b bus.Bus,
	returnIfIdle bool,
	pluginID, instanceKey string,
	valDisposeCallback func(),
) (LoadPluginValue, directive.Instance, directive.Reference, error) {
	return bus.ExecWaitValue[RunningPlugin](
		ctx,
		b,
		NewLoadPluginInstanced(pluginID, instanceKey),
		bus.ReturnIfIdle(returnIfIdle),
		valDisposeCallback,
		nil,
	)
}

// ExPluginLoadWaitClient calls LoadPlugin and returns the rpc client to be set.
// If returnIfIdle is set, it returns nil, nil, nil when the directive becomes idle.
func ExPluginLoadWaitClient(
	ctx context.Context,
	b bus.Bus,
	pluginID string,
	valDisposeCb func(),
) (srpc.Client, directive.Reference, error) {
	rp, _, rpRef, err := ExLoadPlugin(ctx, b, false, pluginID, valDisposeCb)
	return waitRunningPluginClient(rp, rpRef, err)
}

// ExPluginLoadInstancedWaitClient calls LoadPlugin with an instance key and
// returns the rpc client to be set.
func ExPluginLoadInstancedWaitClient(
	ctx context.Context,
	b bus.Bus,
	pluginID, instanceKey string,
	valDisposeCb func(),
) (srpc.Client, directive.Reference, error) {
	rp, _, rpRef, err := ExLoadPluginInstanced(ctx, b, false, pluginID, instanceKey, valDisposeCb)
	return waitRunningPluginClient(rp, rpRef, err)
}

// ExPluginLoadAtManifestWaitClient loads exactly one retained manifest.
// A missing or ineligible artifact returns an error rather than selecting newer code.
func ExPluginLoadAtManifestWaitClient(ctx context.Context, b bus.Bus, pluginID, manifestRoot string) (srpc.Client, directive.Reference, error) {
	rp, _, ref, err := bus.ExecWaitValue[RunningPlugin](
		ctx, b, NewLoadPluginAtManifest(pluginID, "", manifestRoot),
		bus.ReturnIfIdle(true), nil, nil,
	)
	if err == nil && rp == nil {
		err = errors.Errorf("plugin %s: exact manifest %s is unavailable", pluginID, manifestRoot)
	}
	return waitRunningPluginClient(rp, ref, err)
}

func waitRunningPluginClient(
	rp RunningPlugin,
	rpRef directive.Reference,
	err error,
) (srpc.Client, directive.Reference, error) {
	if err != nil || rp == nil {
		if rpRef != nil {
			rpRef.Release()
		}
		return nil, nil, err
	}

	return rp.GetRpcClient(), rpRef, nil
}

// ExPluginLoadAccess calls LoadPlugin and passes the running plugin to cb.
//
// The callback is canceled and restarted if the plugin becomes invalid. Its
// context is canceled when the plugin value changes, and it should return
// context.Canceled in that case. If the callback returns nil, so does
// ExPluginLoadAccess.
func ExPluginLoadAccess(
	ctx context.Context,
	b bus.Bus,
	pluginID string,
	cb func(ctx context.Context, rp RunningPlugin) error,
) error {
	// Create a state routine container that deduplicates identical values.
	routineCtr := routine.NewStateRoutineContainer(
		func(t1, t2 LoadPluginValue) bool { return t1 == t2 },
	)

	// Start a one-off watch for the plugin load directive on the bus.
	di, dirRef, err := bus.ExecOneOffWatchRoutine(
		ctx,
		routineCtr,
		b,
		NewLoadPlugin(pluginID),
	)
	if err != nil {
		return err
	}
	defer dirRef.Release()

	// Forward the first resolution error to the wait channel when idle.
	errCh := make(chan error, 1)
	defer di.AddIdleCallback(func(isIdle bool, resErrs []error) {
		if !isIdle {
			return
		}
		for _, err := range resErrs {
			if err != nil {
				select {
				case errCh <- err:
				default:
				}
				return
			}
		}
	})()

	// Run the callback as the state routine until the watch exits.
	routineCtr.SetContext(ctx, true)
	routineCtr.SetStateRoutine(cb)
	return routineCtr.WaitExited(ctx, false, errCh)
}

// ExPluginLoadAccessClient calls LoadPlugin and passes the rpc client to cb.
//
// The callback is canceled and restarted if the client becomes invalid. Its
// context is canceled when the client value changes, and it should return
// context.Canceled in that case. If the callback returns nil, so does
// ExPluginLoadAccessClient.
func ExPluginLoadAccessClient(
	ctx context.Context,
	b bus.Bus,
	pluginID string,
	cb func(ctx context.Context, c srpc.Client) error,
) error {
	return ExPluginLoadAccess(ctx, b, pluginID, func(ctx context.Context, rp RunningPlugin) error {
		return cb(ctx, rp.GetRpcClient())
	})
}

// Validate checks that the directive's values look correct.
func (d *loadPlugin) Validate() error {
	if d.pluginID == "" {
		return ErrEmptyPluginID
	}

	if d.manifestRoot != "" {
		var root hash.Hash
		if err := root.ParseFromB58(d.manifestRoot); err != nil {
			return err
		}
		return root.Validate()
	}
	for _, ref := range d.manifests {
		if err := ref.Validate(); err != nil {
			return err
		}
		if ref.GetMeta().GetManifestId() != d.pluginID {
			return errors.New("selected manifest does not match plugin ID")
		}
	}
	return nil
}

// GetValueOptions returns options relating to value handling.
func (d *loadPlugin) GetValueOptions() directive.ValueOptions {
	return directive.ValueOptions{}
}

// LoadPluginID returns the plugin ID.
func (d *loadPlugin) LoadPluginID() string {
	return d.pluginID
}

// LoadPluginInstanceKey returns the instance key.
func (d *loadPlugin) LoadPluginInstanceKey() string {
	return d.instanceKey
}

// LoadPluginManifestRoot returns the exact manifest root, or empty for latest selection.
func (d *loadPlugin) LoadPluginManifestRoot() string {
	return d.manifestRoot
}

// LoadPluginManifests returns immutable installation choices, newest first.
func (d *loadPlugin) LoadPluginManifests() []*manifest.ManifestRef {
	return d.manifests
}

// LoadPluginServeAssets reports whether the caller serves the plugin's files.
func (d *loadPlugin) LoadPluginServeAssets() bool {
	return d.serveAssets
}

// IsEquivalent checks if the other directive is equivalent. If two
// directives are equivalent, and the new directive does not superceed the
// old, then the new directive will be merged (de-duplicated) into the old.
func (d *loadPlugin) IsEquivalent(other directive.Directive) bool {
	od, ok := other.(LoadPlugin)
	if !ok {
		return false
	}
	return d.LoadPluginID() == od.LoadPluginID() &&
		d.LoadPluginInstanceKey() == od.LoadPluginInstanceKey() &&
		d.LoadPluginManifestRoot() == od.LoadPluginManifestRoot() &&
		d.LoadPluginServeAssets() == od.LoadPluginServeAssets() &&
		slices.EqualFunc(d.LoadPluginManifests(), od.LoadPluginManifests(), (*manifest.ManifestRef).EqualVT)
}

// GetName returns the directive's type name.
func (d *loadPlugin) GetName() string {
	return "LoadPlugin"
}

// GetDebugVals returns the directive arguments for display. They are not
// necessarily unique.
func (d *loadPlugin) GetDebugVals() directive.DebugValues {
	// Collect the plugin identity and manifest fields for display.
	vals := directive.DebugValues{}
	vals["plugin-id"] = []string{d.LoadPluginID()}
	if ik := d.LoadPluginInstanceKey(); ik != "" {
		vals["instance-key"] = []string{ik}
	}
	if root := d.manifestRoot; root != "" {
		vals["manifest-root"] = []string{root}
	}
	if len(d.manifests) != 0 {
		vals["installed-root"] = []string{d.manifests[0].GetManifestRef().GetRootRef().GetHash().MarshalString()}
	}
	if d.serveAssets {
		vals["serve-assets"] = []string{"true"}
	}
	return vals
}

// _ is a type assertion
var (
	_ LoadPlugin                   = (*loadPlugin)(nil)
	_ directive.DirectiveWithEquiv = (*loadPlugin)(nil)
)