package plugins import ( "bytes" "context" "encoding/json" "errors" "fmt" "log" "net/http" "os" "path/filepath" "sort" "strings" "sync" "time" ) // secretMask is what a set secret value is echoed back as. On save, a field that // still equals the mask is left unchanged (mirrors the pb-config password flow). const secretMask = "••••••••" // record is the persisted state for one plugin. For builtins, Kind/BaseURL are // omitted (the descriptor comes from the registry); external plugins set them. type record struct { Kind string `json:"kind,omitempty"` BaseURL string `json:"baseURL,omitempty"` Provider string `json:"provider,omitempty"` Enabled bool `json:"enabled"` Config map[string]string `json:"config,omitempty"` } // View is the plugin shape returned to the panel (secrets masked). type View struct { Descriptor Enabled bool `json:"enabled"` Config map[string]string `json:"config"` BaseURL string `json:"baseURL,omitempty"` Health *Health `json:"health,omitempty"` } // Manager owns the plugin registry, persisted state, and live instances. type Manager struct { path string mu sync.Mutex factories map[string]Factory records map[string]*record live map[string]Plugin health map[string]*Health client *http.Client } // NewManager builds a Manager backed by the JSON state file at path. func NewManager(path string) *Manager { return &Manager{ path: path, factories: builtinFactories(), records: map[string]*record{}, live: map[string]Plugin{}, health: map[string]*Health{}, client: &http.Client{Timeout: 12 * time.Second}, } } // Load reads the state file and initialises every enabled plugin. A missing file // is fine (no plugins configured yet). func (m *Manager) Load() error { m.mu.Lock() defer m.mu.Unlock() data, err := os.ReadFile(m.path) switch { case err == nil: if err := m.decodeLocked(data); err != nil { return err } case errors.Is(err, os.ErrNotExist): // No state file yet — first boot, nothing configured. default: return err } ctx := context.Background() for name, rec := range m.records { if !rec.Enabled { continue } p := construct(name, m.factories[name], rec) if p == nil { log.Printf("plugins: cannot construct %q (unknown builtin?)", name) continue } if err := p.Init(ctx, rec.Config); err != nil { log.Printf("plugins: init %q failed: %v", name, err) continue } m.live[name] = p } return nil } // decodeLocked parses the state file into m.records. It is deliberately strict // about two shapes that would otherwise take the server down or quietly destroy // state: // // - An empty file, or a literal "null", decodes to a nil map. Assigning that // to m.records makes the next save panic with "assignment to entry in nil // map", so both are treated as "nothing configured" instead. // - A null entry ({"toyota": null}) leaves a nil *record that the enable loop // in Load would dereference. Those entries are dropped. // // Content that does not parse at all is moved aside rather than left in place: // the server carries on with no plugins configured, and the next save would // otherwise overwrite the unreadable file and take every setting in it along. func (m *Manager) decodeLocked(data []byte) error { if len(bytes.TrimSpace(data)) == 0 { return nil } var recs map[string]*record if err := json.Unmarshal(data, &recs); err != nil { backup := m.path + ".corrupt" if renameErr := os.Rename(m.path, backup); renameErr != nil { return fmt.Errorf("plugin state file %s is unreadable (%v) and could not be set aside: %v", m.path, err, renameErr) } return fmt.Errorf("plugin state file %s was unreadable (%v); moved it to %s and started with no plugins configured", m.path, err, backup) } for name, rec := range recs { if rec == nil { delete(recs, name) } } if recs != nil { m.records = recs } return nil } // CheckWritable reports whether the state file can actually be written, by // creating and removing a temporary file beside it. Worth calling at startup: // an unwritable state directory (a root-owned bind mount under an unprivileged // process, or a volume left over from an image that ran as root) otherwise // stays invisible until a restart brings every plugin back disabled. func (m *Manager) CheckWritable() error { f, err := os.CreateTemp(filepath.Dir(m.path), ".plugins-writecheck-*") if err != nil { return fmt.Errorf("%w: %v", errPersist, err) } name := f.Name() _ = f.Close() _ = os.Remove(name) return nil } // construct builds a plugin instance from a builtin factory or an external record. func construct(name string, f Factory, rec *record) Plugin { if f != nil { return f() } if rec != nil && rec.Kind == KindExternal { return newExternalPlugin(name, rec.BaseURL, rec.Provider) } return nil } // descriptorFor returns a plugin's descriptor without needing a live instance. func (m *Manager) descriptorFor(name string, rec *record) Descriptor { if p := m.live[name]; p != nil { return p.Descriptor() } if f := m.factories[name]; f != nil { return f().Descriptor() } if rec != nil && rec.Kind == KindExternal { return newExternalPlugin(name, rec.BaseURL, rec.Provider).Descriptor() } return Descriptor{Name: name} } // maskConfig echoes config back with secret fields masked when set. func maskConfig(d Descriptor, cfg map[string]string) map[string]string { out := map[string]string{} for k, v := range cfg { out[k] = v } for _, f := range d.ConfigFields { if f.Secret && out[f.Key] != "" { out[f.Key] = secretMask } } return out } // List returns every known plugin (registry ∪ persisted), sorted by name. func (m *Manager) List() []View { m.mu.Lock() defer m.mu.Unlock() names := map[string]bool{} for n := range m.factories { names[n] = true } for n := range m.records { names[n] = true } out := make([]View, 0, len(names)) for name := range names { rec := m.records[name] d := m.descriptorFor(name, rec) v := View{Descriptor: d, Health: m.health[name]} if rec != nil { v.Enabled = rec.Enabled v.BaseURL = rec.BaseURL v.Config = maskConfig(d, rec.Config) } else { v.Config = map[string]string{} } out = append(out, v) } sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) return out } // Get returns a single plugin view (ok=false when unknown). func (m *Manager) Get(name string) (View, bool) { for _, v := range m.List() { if v.Name == name { return v, true } } return View{}, false } // Upsert enables/disables a plugin and merges its config, then (re)initialises or // shuts down the live instance to match. Secrets left at the mask are preserved. func (m *Manager) Upsert(ctx context.Context, name string, enabled bool, incoming map[string]string) (View, error) { m.mu.Lock() _, isBuiltin := m.factories[name] rec := m.records[name] if !isBuiltin && (rec == nil || rec.Kind != KindExternal) { m.mu.Unlock() return View{}, errUnknown } isNew := rec == nil if isNew { rec = &record{} m.records[name] = rec } d := m.descriptorFor(name, rec) merged := map[string]string{} for k, v := range rec.Config { merged[k] = v } // Apply incoming values, honouring the secret-mask keep-current rule. secretKeys := map[string]bool{} for _, f := range d.ConfigFields { if f.Secret { secretKeys[f.Key] = true } } for k, v := range incoming { if secretKeys[k] && v == secretMask { continue // keep existing secret } merged[k] = strings.TrimSpace(v) } // Validate required fields when enabling. if enabled { for _, f := range d.ConfigFields { if f.Required && merged[f.Key] == "" { if isNew { delete(m.records, name) // don't leave a blank record behind } m.mu.Unlock() return View{}, errors.New("missing required setting: " + f.Label) } } } prevEnabled, prevConfig := rec.Enabled, rec.Config rec.Enabled = enabled rec.Config = merged if err := m.persistLocked(); err != nil { // Roll back, so the panel shows what is actually on disk. Keeping the // change in memory is what made an unwritable state file look like a // successful save — right up until the next restart brought it back // disabled. if isNew { delete(m.records, name) } else { rec.Enabled, rec.Config = prevEnabled, prevConfig } m.mu.Unlock() return View{}, err } // Reconcile the live instance. if old := m.live[name]; old != nil { _ = old.Shutdown(ctx) delete(m.live, name) } var initErr error if enabled { p := construct(name, m.factories[name], rec) if p != nil { if err := p.Init(ctx, merged); err != nil { initErr = err } else { m.live[name] = p } } } m.mu.Unlock() v, _ := m.Get(name) return v, initErr } // RegisterExternal adds a new external (remote HTTP) plugin at runtime — the // "add a plugin without a rebuild" path. It starts disabled. func (m *Manager) RegisterExternal(name, baseURL, provider string) error { name = strings.TrimSpace(name) baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") if name == "" || baseURL == "" { return errors.New("name and baseURL are required") } if !strings.HasPrefix(baseURL, "http://") && !strings.HasPrefix(baseURL, "https://") { baseURL = "http://" + baseURL } m.mu.Lock() defer m.mu.Unlock() if _, dup := m.factories[name]; dup { return errors.New("a builtin plugin already uses that name") } if _, dup := m.records[name]; dup { return errors.New("a plugin with that name already exists") } m.records[name] = &record{Kind: KindExternal, BaseURL: baseURL, Provider: provider} if err := m.persistLocked(); err != nil { delete(m.records, name) return err } return nil } // Remove deletes an external plugin registration. Builtins can only be disabled. func (m *Manager) Remove(ctx context.Context, name string) error { m.mu.Lock() defer m.mu.Unlock() rec := m.records[name] if rec == nil || rec.Kind != KindExternal { return errors.New("only external plugins can be removed") } // Persist before tearing the instance down, so a failed write leaves a // still-registered plugin still running rather than a half-removed one. delete(m.records, name) if err := m.persistLocked(); err != nil { m.records[name] = rec return err } if p := m.live[name]; p != nil { _ = p.Shutdown(ctx) delete(m.live, name) } delete(m.health, name) return nil } // HealthCheck probes a plugin now, building a transient instance if it is not // currently live (so disabled plugins can still be tested). Result is cached. func (m *Manager) HealthCheck(ctx context.Context, name string) (Health, error) { m.mu.Lock() p := m.live[name] transient := false var cfg map[string]string if p == nil { rec := m.records[name] if rec != nil { cfg = rec.Config } p = construct(name, m.factories[name], rec) transient = true } m.mu.Unlock() if p == nil { return Health{}, errUnknown } if transient { _ = p.Init(ctx, cfg) defer func() { _ = p.Shutdown(context.Background()) }() } h := p.HealthCheck(ctx) m.mu.Lock() hc := h m.health[name] = &hc m.mu.Unlock() return h, nil } // HealthCheckWith probes a plugin against a caller-resolved config rather than // the stored global config. It builds a transient instance, Inits it with cfg, // probes, and tears it down — so a per-user cascade (see internal/api/ // integrations.go) can health-check under the credentials in force for that // caller without disturbing the global instance or its cached health. func (m *Manager) HealthCheckWith(ctx context.Context, name string, cfg map[string]string) (Health, error) { m.mu.Lock() rec := m.records[name] p := construct(name, m.factories[name], rec) m.mu.Unlock() if p == nil { return Health{}, errUnknown } _ = p.Init(ctx, cfg) defer func() { _ = p.Shutdown(context.Background()) }() return p.HealthCheck(ctx), nil } // InvokeWith runs a capability against a caller-resolved config. Like // HealthCheckWith, it uses a transient instance Inited with cfg so per-user // credentials drive the call. Returns the plugin's raw JSON result. func (m *Manager) InvokeWith(ctx context.Context, name string, cfg map[string]string, action string, payload json.RawMessage) (json.RawMessage, error) { m.mu.Lock() rec := m.records[name] p := construct(name, m.factories[name], rec) m.mu.Unlock() if p == nil { return nil, errUnknown } _ = p.Init(ctx, cfg) defer func() { _ = p.Shutdown(context.Background()) }() return p.Invoke(ctx, action, payload) } // BatchCall is one capability invocation inside an InvokeBatchWith request. type BatchCall struct { ID string // caller-chosen id, echoed back on the result Action string // capability id Params json.RawMessage // action params; may be nil } // BatchResult is the outcome of one BatchCall. Exactly one of Result/Err is set. type BatchResult struct { ID string Result json.RawMessage Err error } // batchConcurrency caps how many calls of one batch are in flight at once, so a // snapshot of a whole vehicle doesn't arrive at the upstream as a burst. const batchConcurrency = 4 // InvokeBatchWith runs several capabilities against one caller-resolved config, // sharing a single transient instance. A connector that authenticates lazily // (Toyota's OAuth login on first request) would otherwise repeat that login for // every action, because InvokeWith builds and tears down an instance per call; // sharing the instance logs in once for the whole batch. // // Calls run concurrently, so a plugin's Invoke must be safe for concurrent use — // which the contract already implies, since the live instance is shared by every // HTTP request. Results come back in request order, each carrying its own error; // a non-nil error return means the batch never started (unknown plugin). func (m *Manager) InvokeBatchWith(ctx context.Context, name string, cfg map[string]string, calls []BatchCall) ([]BatchResult, error) { m.mu.Lock() rec := m.records[name] p := construct(name, m.factories[name], rec) m.mu.Unlock() if p == nil { return nil, errUnknown } _ = p.Init(ctx, cfg) defer func() { _ = p.Shutdown(context.Background()) }() out := make([]BatchResult, len(calls)) sem := make(chan struct{}, batchConcurrency) var wg sync.WaitGroup for i, c := range calls { wg.Add(1) go func(i int, c BatchCall) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() res, err := p.Invoke(ctx, c.Action, c.Params) out[i] = BatchResult{ID: c.ID, Result: res, Err: err} }(i, c) } wg.Wait() return out, nil } // RawConfig returns a copy of a plugin's stored (global) config and its enabled // flag. ok is false for an unknown plugin. This is the top layer (L1) of the // per-user cascade: the config a superadmin set in the panel, which lower layers // inherit blank fields from. Secrets are returned in clear — callers must mask // before returning anything to a client. func (m *Manager) RawConfig(name string) (cfg map[string]string, enabled, ok bool) { m.mu.Lock() defer m.mu.Unlock() _, isBuiltin := m.factories[name] rec := m.records[name] if !isBuiltin && rec == nil { return nil, false, false } out := map[string]string{} if rec != nil { for k, v := range rec.Config { out[k] = v } enabled = rec.Enabled } return out, enabled, true } // Shutdown tears down every live plugin instance. Wire into graceful shutdown. func (m *Manager) Shutdown(ctx context.Context) { m.mu.Lock() defer m.mu.Unlock() for name, p := range m.live { _ = p.Shutdown(ctx) delete(m.live, name) } } // persistLocked writes the state file. Caller must hold m.mu. // // The write goes to a temporary file in the same directory, is flushed, and is // then renamed over the target. A truncating write in place can be interrupted // (crash, container stop, full disk) and leave a half-written plugins.json that // fails to parse on the next boot — which surfaces as every plugin coming back // disabled. Every failure is wrapped in errPersist so callers can tell "your // change was not saved" apart from "saved, but the plugin failed to start". func (m *Manager) persistLocked() error { data, err := json.MarshalIndent(m.records, "", " ") if err != nil { return fmt.Errorf("%w: %v", errPersist, err) } data = append(data, '\n') tmp, err := os.CreateTemp(filepath.Dir(m.path), ".plugins-*.json") if err != nil { return fmt.Errorf("%w: %v", errPersist, err) } tmpName := tmp.Name() defer func() { if tmpName != "" { _ = os.Remove(tmpName) } }() if _, err := tmp.Write(data); err != nil { _ = tmp.Close() return fmt.Errorf("%w: %v", errPersist, err) } if err := tmp.Sync(); err != nil { _ = tmp.Close() return fmt.Errorf("%w: %v", errPersist, err) } if err := tmp.Close(); err != nil { return fmt.Errorf("%w: %v", errPersist, err) } if err := os.Rename(tmpName, m.path); err != nil { return fmt.Errorf("%w: %v", errPersist, err) } tmpName = "" // renamed into place; nothing left to clean up return nil } var ( errUnknown = errors.New("unknown plugin") errPersist = errors.New("plugin state could not be saved") ) // IsUnknown reports whether err came from addressing a plugin that doesn't exist. func IsUnknown(err error) bool { return errors.Is(err, errUnknown) } // IsPersist reports whether err means the change never reached the state file. // Such a change has been rolled back in memory: it must be reported as a // failure, or the caller sees a save that silently vanishes on the next restart. func IsPersist(err error) bool { return errors.Is(err, errPersist) }