overlord/Overlord-Client/cmd/agent/plugins/manager.go

package plugins

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"sync"

	"overlord-client/cmd/agent/wire"
)

type Manager struct {
	mu      sync.Mutex
	plugins map[string]*pluginInstance
	pending map[string]*pendingBundle
	writer  wire.Writer
	host    HostInfo
}

type pluginInstance struct {
	id       string
	manifest PluginManifest
	native   NativePlugin
}

type pendingBundle struct {
	manifest    PluginManifest
	totalSize   int
	totalChunks int
	chunks      map[int][]byte
	received    int
	receivedSz  int
}

func NewManager(writer wire.Writer, host HostInfo) *Manager {
	return &Manager{
		plugins: make(map[string]*pluginInstance),
		pending: make(map[string]*pendingBundle),
		writer:  writer,
		host:    host,
	}
}

func (m *Manager) StartBundle(manifest PluginManifest, totalSize int, totalChunks int) error {
	if manifest.ID == "" {
		return errors.New("missing plugin id")
	}
	if totalChunks <= 0 {
		return errors.New("invalid total chunks")
	}
	if totalSize <= 0 {
		return errors.New("invalid total size")
	}
	if totalChunks > 10000 {
		return errors.New("too many chunks")
	}

	m.mu.Lock()
	m.pending[manifest.ID] = &pendingBundle{
		manifest:    manifest,
		totalSize:   totalSize,
		totalChunks: totalChunks,
		chunks:      make(map[int][]byte),
	}
	m.mu.Unlock()
	return nil
}

func (m *Manager) AddChunk(pluginId string, index int, data []byte) error {
	if pluginId == "" {
		return errors.New("missing plugin id")
	}
	if index < 0 {
		return errors.New("invalid chunk index")
	}

	m.mu.Lock()
	b := m.pending[pluginId]
	if b == nil {
		m.mu.Unlock()
		return errors.New("bundle not initialized")
	}
	if _, exists := b.chunks[index]; !exists {
		b.chunks[index] = data
		b.received++
		b.receivedSz += len(data)
	}
	m.mu.Unlock()
	return nil
}

func (m *Manager) FinalizeBundle(ctx context.Context, pluginId string) error {
	m.mu.Lock()
	b := m.pending[pluginId]
	if b == nil {
		m.mu.Unlock()
		return errors.New("bundle not initialized")
	}
	if b.received < b.totalChunks {
		m.mu.Unlock()
		return errors.New("bundle incomplete")
	}

	chunks := make([][]byte, b.totalChunks)
	for i := 0; i < b.totalChunks; i++ {
		chunks[i] = b.chunks[i]
		if chunks[i] == nil {
			m.mu.Unlock()
			return errors.New("missing chunk")
		}
	}
	manifest := b.manifest
	delete(m.pending, pluginId)
	m.mu.Unlock()

	combined := make([]byte, 0, b.totalSize)
	for _, part := range chunks {
		combined = append(combined, part...)
	}
	return m.Load(ctx, manifest, combined)
}

func (m *Manager) Load(ctx context.Context, manifest PluginManifest, binary []byte) error {
	if len(binary) == 0 {
		return errors.New("empty plugin binary")
	}
	pluginID := manifest.ID
	if pluginID == "" {
		return errors.New("missing plugin id")
	}

	m.mu.Lock()
	if existing, ok := m.plugins[pluginID]; ok {
		existing.native.Close()
		delete(m.plugins, pluginID)
	}
	m.mu.Unlock()

	np, err := loadNativePlugin(binary)
	if err != nil {
		return err
	}

	pi := &pluginInstance{
		id:       pluginID,
		manifest: manifest,
		native:   np,
	}

	send := func(event string, payload []byte) {
		var payloadVal interface{}
		if len(payload) > 0 {
			var parsed interface{}
			if json.Unmarshal(payload, &parsed) == nil {
				payloadVal = parsed
			} else {
				payloadVal = string(payload)
			}
		}
		err := wire.WriteMsg(context.Background(), m.writer, wire.PluginEvent{
			Type:     "plugin_event",
			PluginID: pluginID,
			Event:    event,
			Payload:  payloadVal,
		})
		if err != nil {
			log.Printf("[plugin] %s send event error: %v", pluginID, err)
		}
	}

	hostJSON, err := json.Marshal(m.host)
	if err != nil {
		np.Close()
		return err
	}

	if err := np.Load(send, hostJSON); err != nil {
		np.Close()
		return err
	}

	m.mu.Lock()
	m.plugins[pluginID] = pi
	m.mu.Unlock()

	rt := np.Runtime()
	freeable := rt != "go"
	log.Printf("[plugin] loaded %s (native, runtime=%s, freeable=%v)", pluginID, rt, freeable)
	return nil
}

func (m *Manager) Dispatch(ctx context.Context, pluginId, event string, payload interface{}) error {
	m.mu.Lock()
	pi := m.plugins[pluginId]
	m.mu.Unlock()
	if pi == nil {
		return nil
	}

	data, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	return pi.native.Event(event, data)
}

func (m *Manager) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, pi := range m.plugins {
		rt := pi.native.Runtime()
		pi.native.Close()
		delete(m.plugins, id)
		if rt != "go" {
			log.Printf("[plugin] unloaded %s (runtime=%s, memory freed)", id, rt)
		} else {
			log.Printf("[plugin] unloaded %s (runtime=go, memory leaked — see golang/go#11100)", id)
		}
	}
}

func (m *Manager) Unload(pluginId string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if pi, ok := m.plugins[pluginId]; ok {
		rt := pi.native.Runtime()
		pi.native.Close()
		delete(m.plugins, pluginId)
		if rt != "go" {
			log.Printf("[plugin] unloaded %s (runtime=%s, memory freed)", pluginId, rt)
		} else {
			log.Printf("[plugin] unloaded %s (runtime=go, memory leaked — see golang/go#11100)", pluginId)
		}
	}
}