controller/task_plugin.go
Rama main · réplica de lectura del repositorio original.
package controller
import (
"context"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
"maps"
"net/http"
"net/url"
"sort"
"strings"
"sync"
"time"
"github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/logger"
"github.com/QuantumNous/new-api/model"
"github.com/QuantumNous/new-api/pkg/jsplugin"
"github.com/QuantumNous/new-api/plugins"
"github.com/QuantumNous/new-api/setting"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
const maxTaskPluginSourceBytes = 8 << 20
func taskPluginCompileError(c *gin.Context, err error) {
var unknownField *jsplugin.UnknownMetaFieldError
if errors.As(err, &unknownField) {
common.ApiErrorT(c, "Plugin metadata contains an unknown field \"{{field}}\". If this plugin was downloaded from the official marketplace, it may require a newer version of new-api. Try updating new-api and installing the plugin again.", map[string]any{"field": unknownField.Field})
return
}
common.ApiError(c, err)
}
type taskPluginUploadRequest struct {
Source string `json:"source" binding:"required"`
Enabled *bool `json:"enabled"`
Remark string `json:"remark"`
SourceSha256 string `json:"sourceSha256"`
// Icon carries the sidecar icon.svg / icon.png as a data URI. It is optional
// and stored separately from the source so the JavaScript stays readable.
Icon string `json:"icon"`
}
func UploadTaskPlugin(c *gin.Context) {
var request taskPluginUploadRequest
if err := c.ShouldBindJSON(&request); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
if len(request.Source) > maxTaskPluginSourceBytes {
common.ApiErrorMsg(c, "plugin source exceeds 8 MiB")
return
}
if expected := strings.TrimSpace(request.SourceSha256); expected != "" {
actual := fmt.Sprintf("%x", sha256.Sum256([]byte(request.Source)))
if !strings.EqualFold(actual, expected) {
common.ApiErrorMsg(c, "plugin source sha256 mismatch")
return
}
}
temporary := jsplugin.NewRegistry()
loaded, err := temporary.Register(request.Source, jsplugin.Options{})
if err != nil {
taskPluginCompileError(c, err)
return
}
if err = jsplugin.ValidateV1Meta(loaded.Meta); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
icon := strings.TrimSpace(request.Icon)
if icon != "" {
if _, _, err = jsplugin.DecodeIconDataURI(icon); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
}
enabled := true
if request.Enabled != nil {
enabled = *request.Enabled
}
if enabled {
if err = jsplugin.PreflightRoutingConflict(jsplugin.DefaultRegistry.Generation(), loaded); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
} else {
// A plugin stored switched off is not checked, unless the upload
// switches off the version serving its key: the factory built-in
// then serves again.
active, activeErr := model.GetTaskPluginVersion(loaded.Meta.Key, "")
if activeErr != nil && !errors.Is(activeErr, gorm.ErrRecordNotFound) {
common.ApiError(c, activeErr)
return
}
if activeErr == nil && active.Version == loaded.Meta.Version && active.Enabled && !setting.IsTaskPluginFactoryDisabled(loaded.Meta.Key) {
if rejectTaskPluginRoutingConflict(c, loaded.Meta.Key, nil) {
return
}
}
}
plugin := model.TaskPlugin{
Key: loaded.Meta.Key, APIVersion: loaded.Meta.APIVersion, Version: loaded.Meta.Version,
Source: model.LongText(request.Source), SourceHash: fmt.Sprintf("%x", sha256.Sum256([]byte(request.Source))),
Icon: model.LongText(icon), Enabled: enabled, Remark: request.Remark,
}
if err = model.SaveTaskPlugin(&plugin); err != nil {
common.ApiError(c, err)
return
}
if err = syncTaskPluginsOnceContext(c.Request.Context()); err != nil {
common.ApiError(c, err)
return
}
detail := taskPluginDetail{Plugin: &plugin, Meta: loaded.Meta, Source: string(plugin.Source), Layer: "override", HasIcon: plugin.HasIcon(), SkippedRoutes: taskPluginSkippedRoutes(plugin.Key, plugin.Version)}
if plugin.Active && plugin.Enabled {
detail.PricingIssues = taskPluginPricingIssues(plugin.Key)
}
common.ApiSuccess(c, detail)
}
func GetTaskPluginVersions(c *gin.Context) {
plugins, err := model.ListTaskPluginVersions(c.Param("key"))
if err != nil {
common.ApiError(c, err)
return
}
common.ApiSuccess(c, plugins)
}
type taskPluginListItem struct {
Meta jsplugin.Meta `json:"meta"`
Source string `json:"source"`
Enabled bool `json:"enabled"`
Active bool `json:"active"`
SourceHash string `json:"source_hash"`
HasIcon bool `json:"has_icon"`
Remark string `json:"remark"`
RuntimeStatus string `json:"runtime_status"`
RuntimeError string `json:"runtime_error,omitempty"`
FactoryMeta *jsplugin.Meta `json:"factory_meta,omitempty"`
ChannelCount int `json:"channel_count"`
InFlightCount int64 `json:"in_flight_count"`
SkippedRoutes []string `json:"skipped_routes"`
}
type taskPluginRebuildOutcome struct {
Status string `json:"status"`
AttemptedAt time.Time `json:"attempted_at"`
Generation uint64 `json:"generation"`
DatabaseRevision string `json:"database_revision,omitempty"`
PluginErrorCount int `json:"plugin_error_count"`
Error string `json:"error,omitempty"`
}
type taskPluginRuntimeStatus struct {
CurrentGeneration uint64 `json:"current_generation"`
GenerationPublishedAt time.Time `json:"generation_published_at"`
DatabaseRevision string `json:"database_revision"`
DatabaseError string `json:"database_error,omitempty"`
LastRebuild taskPluginRebuildOutcome `json:"last_rebuild"`
PluginErrors map[string]string `json:"plugin_errors"`
}
func ListTaskPlugins(c *gin.Context) {
databasePlugins, err := model.ListTaskPlugins()
if err != nil {
common.ApiError(c, err)
return
}
snapshot := jsplugin.DefaultRegistry.Snapshot()
factory := make(map[string]jsplugin.Meta, len(snapshot.Factory))
override := make(map[string]jsplugin.Meta, len(snapshot.Override))
for _, meta := range snapshot.Factory {
factory[meta.Key] = meta
}
for _, meta := range snapshot.Override {
override[meta.Key] = meta
}
activeRows := make(map[string]model.TaskPlugin)
keys := make(map[string]struct{}, len(factory)+len(databasePlugins))
for key := range factory {
keys[key] = struct{}{}
}
for _, plugin := range databasePlugins {
keys[plugin.Key] = struct{}{}
if plugin.Active {
activeRows[plugin.Key] = plugin
}
}
runtimeErrors := jsplugin.DefaultRegistry.RoutingErrors()
taskPluginSyncState.Lock()
maps.Copy(runtimeErrors, taskPluginSyncState.errors)
taskPluginSyncState.Unlock()
items := make([]taskPluginListItem, 0, len(keys))
for key := range keys {
factoryMeta, hasFactory := factory[key]
row, hasOverride := activeRows[key]
item := taskPluginListItem{Enabled: true, Active: true, RuntimeStatus: "registered"}
if hasOverride {
item.Source = "override"
if hasFactory {
item.Source = "override_over_factory"
factoryCopy := factoryMeta
item.FactoryMeta = &factoryCopy
}
item.Meta = jsplugin.Meta{Key: row.Key, Version: row.Version, APIVersion: row.APIVersion}
if compiled, compileErr := jsplugin.NewRegistry().Register(string(row.Source), jsplugin.Options{Key: row.Key, Version: row.Version}); compileErr == nil {
item.Meta = compiled.Meta
}
item.Enabled = row.Enabled
item.Active = row.Active
item.SourceHash = row.SourceHash
item.HasIcon = row.HasIcon()
item.Remark = row.Remark
if message := runtimeErrors[key]; message != "" {
item.RuntimeStatus = "compile_failed"
item.RuntimeError = message
} else if runtimeMeta, ok := override[key]; ok {
item.Meta = runtimeMeta
} else if !row.Enabled {
item.RuntimeStatus = "disabled_fallback"
} else {
item.RuntimeStatus = "not_registered"
}
// "disabled_fallback" promises that the built-in still serves. When
// the factory layer is suppressed as well, nothing serves this key.
if item.RuntimeStatus == "disabled_fallback" && hasFactory && setting.IsTaskPluginFactoryDisabled(key) {
item.RuntimeStatus = "disabled"
}
} else {
item.Source = "factory"
item.Meta = factoryMeta
item.Enabled = !setting.IsTaskPluginFactoryDisabled(key)
_, _, item.HasIcon = plugins.Icon(key)
source, sourceErr := plugins.Source(key)
if sourceErr == nil {
item.SourceHash = fmt.Sprintf("%x", sha256.Sum256([]byte(source)))
}
if !item.Enabled {
item.RuntimeStatus = "disabled"
} else if message := runtimeErrors[key]; message != "" {
item.RuntimeStatus = "compile_failed"
item.RuntimeError = message
}
}
if !hasFactory {
channels, inFlight, usageErr := model.GetTaskPluginUsage(key)
if usageErr != nil {
common.ApiError(c, usageErr)
return
}
item.ChannelCount = len(channels)
item.InFlightCount = inFlight
}
item.SkippedRoutes = taskPluginSkippedRoutes(key, item.Meta.Version)
items = append(items, item)
}
sort.Slice(items, func(i, j int) bool {
if items[i].Meta.SortPriority != items[j].Meta.SortPriority {
return items[i].Meta.SortPriority > items[j].Meta.SortPriority
}
return items[i].Meta.Key < items[j].Meta.Key
})
common.ApiSuccess(c, items)
}
func GetTaskPluginRuntime(c *gin.Context) {
routingStatus := jsplugin.DefaultRegistry.RoutingStatus()
pluginErrors := routingStatus.Errors
taskPluginSyncState.Lock()
maps.Copy(pluginErrors, taskPluginSyncState.errors)
lastRebuild := taskPluginSyncState.lastRebuild
lastDatabaseRevision := lastRebuild.DatabaseRevision
taskPluginSyncState.Unlock()
registryRebuild := routingStatus.LastRebuild
if lastRebuild.AttemptedAt.Before(registryRebuild.AttemptedAt) {
lastRebuild = taskPluginRebuildOutcome{
Status: registryRebuild.Status,
AttemptedAt: registryRebuild.AttemptedAt,
Generation: registryRebuild.Generation,
Error: registryRebuild.Error,
}
}
if lastRebuild.Status == "" {
lastRebuild.Status = "never"
}
lastRebuild.PluginErrorCount = len(pluginErrors)
if lastRebuild.Status == "success" && len(pluginErrors) > 0 {
lastRebuild.Status = "partial"
}
status := taskPluginRuntimeStatus{
DatabaseRevision: lastDatabaseRevision,
LastRebuild: lastRebuild,
PluginErrors: pluginErrors,
}
databaseSnapshot, err := model.GetTaskPluginSyncSnapshot()
if err != nil {
status.DatabaseError = "database snapshot unavailable"
} else {
status.DatabaseRevision = databaseSnapshot.Revision
}
if routingStatus.Generation != nil {
status.CurrentGeneration = routingStatus.Generation.Number
status.GenerationPublishedAt = routingStatus.Generation.PublishedAt
}
common.ApiSuccess(c, status)
}
type taskPluginDetail struct {
Plugin *model.TaskPlugin `json:"plugin,omitempty"`
Meta jsplugin.Meta `json:"meta"`
Source string `json:"source"`
Layer string `json:"layer"`
HasIcon bool `json:"has_icon"`
SkippedRoutes []string `json:"skipped_routes"`
// PricingIssues is set by an upload that puts the version into service.
PricingIssues []model.ModelPricingIssue `json:"pricing_issues,omitempty"`
}
// taskPluginSkippedRoutes lists, as "METHOD /path", the declared top-level
// routes the gateway does not register for the plugin version serving key;
// the plugin's /{key} copies still serve them.
func taskPluginSkippedRoutes(key, version string) []string {
skipped := make([]string, 0)
generation := jsplugin.DefaultRegistry.Generation()
plugin, registered := generation.Get(key)
if !registered || plugin.Meta.Version != version {
return skipped
}
for _, route := range generation.SkippedRoutes() {
if route.PluginKey == key {
skipped = append(skipped, route.Route.Method+" "+route.Route.Path)
}
}
return skipped
}
// GetTaskPluginIcon serves a plugin logo as an image. The active override wins
// (or the requested ?version=), then the factory sidecar. Data icons are only
// ever drawn through <img>, and the nosniff header keeps a browser from
// treating an SVG response as a document.
func GetTaskPluginIcon(c *gin.Context) {
key := c.Param("key")
icon := ""
plugin, err := model.GetTaskPluginVersion(key, c.Query("version"))
if err == nil {
icon = string(plugin.Icon)
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
common.ApiError(c, err)
return
}
if icon == "" && c.Query("version") == "" {
icon = plugins.IconDataURI(key)
}
if icon == "" {
c.AbortWithStatus(http.StatusNotFound)
return
}
mediaType, data, err := jsplugin.DecodeIconDataURI(icon)
if err != nil {
c.AbortWithStatus(http.StatusNotFound)
return
}
c.Header("Cache-Control", "private, max-age=3600")
c.Header("X-Content-Type-Options", "nosniff")
c.Header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; sandbox")
c.Data(http.StatusOK, mediaType, data)
}
func GetTaskPlugin(c *gin.Context) {
key := c.Param("key")
version := c.Query("version")
plugin, err := model.GetTaskPluginVersion(key, version)
if err == nil {
loaded, compileErr := jsplugin.NewRegistry().Register(string(plugin.Source), jsplugin.Options{Key: plugin.Key, Version: plugin.Version})
if compileErr != nil {
taskPluginCompileError(c, compileErr)
return
}
common.ApiSuccess(c, taskPluginDetail{Plugin: plugin, Meta: loaded.Meta, Source: string(plugin.Source), Layer: "override", HasIcon: plugin.HasIcon(), SkippedRoutes: taskPluginSkippedRoutes(plugin.Key, plugin.Version)})
return
}
if !errors.Is(err, gorm.ErrRecordNotFound) || version != "" {
common.ApiError(c, err)
return
}
source, err := plugins.Source(key)
if err != nil {
common.ApiErrorMsg(c, "task plugin not found")
return
}
loaded, err := jsplugin.NewRegistry().RegisterFactory(source, jsplugin.Options{Key: key})
if err != nil {
taskPluginCompileError(c, err)
return
}
_, _, hasIcon := plugins.Icon(key)
common.ApiSuccess(c, taskPluginDetail{Meta: loaded.Meta, Source: source, Layer: "factory", HasIcon: hasIcon, SkippedRoutes: taskPluginSkippedRoutes(key, loaded.Meta.Version)})
}
type taskPluginDryRunRequest struct {
Hook string `json:"hook" binding:"required"`
Member string `json:"member"`
Args []json.RawMessage `json:"args"`
}
func DryRunTaskPlugin(c *gin.Context) {
var request taskPluginDryRunRequest
if err := c.ShouldBindJSON(&request); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
detailSource := ""
plugin, err := model.GetTaskPluginVersion(c.Param("key"), "")
if err == nil {
detailSource = string(plugin.Source)
} else if errors.Is(err, gorm.ErrRecordNotFound) {
detailSource, err = plugins.Source(c.Param("key"))
}
if err != nil {
common.ApiErrorMsg(c, "task plugin not found")
return
}
loaded, err := jsplugin.NewRegistry().Register(detailSource, jsplugin.Options{Key: c.Param("key")})
if err != nil {
taskPluginCompileError(c, err)
return
}
args := make([]any, len(request.Args))
for index, raw := range request.Args {
if err = common.Unmarshal(raw, &args[index]); err != nil {
common.ApiErrorMsg(c, fmt.Sprintf("invalid argument %d: %v", index+1, err))
return
}
}
var output any
if request.Member == "" {
output, err = loaded.Engine.Call(context.Background(), request.Hook, args...)
} else {
output, err = loaded.Engine.CallMember(context.Background(), request.Hook, request.Member, args...)
}
if err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
common.ApiSuccess(c, output)
}
func DeleteTaskPluginVersion(c *gin.Context) {
key := c.Param("key")
version := c.Param("version")
plugin, lookupErr := model.GetTaskPluginVersion(key, version)
if lookupErr != nil {
if errors.Is(lookupErr, gorm.ErrRecordNotFound) {
common.ApiErrorMsg(c, "override plugin version not found; factory plugins cannot be deleted")
return
}
common.ApiError(c, lookupErr)
return
}
if plugin.Active {
versions, err := model.ListTaskPluginVersions(key)
if err != nil {
common.ApiError(c, err)
return
}
// Deleting the active version promotes the newest remaining one, in
// the order model.DeleteTaskPluginVersion uses. When that one is
// switched off or none is left, a version that was serving hands the
// key back to the factory built-in.
var promoted *model.TaskPlugin
for i := range versions {
if versions[i].Version != version {
promoted = &versions[i]
break
}
}
switch {
case promoted != nil && promoted.Enabled:
if rejectTaskPluginRoutingConflict(c, key, promoted) {
return
}
case plugin.Enabled && !setting.IsTaskPluginFactoryDisabled(key):
if rejectTaskPluginRoutingConflict(c, key, nil) {
return
}
}
}
if plugin.Active && !taskPluginHasFactory(key) {
channels, inFlight, usageErr := model.GetTaskPluginUsage(key)
if usageErr != nil {
common.ApiError(c, usageErr)
return
}
if (len(channels) > 0 || inFlight > 0) && c.Query("force") != "true" {
c.JSON(200, gin.H{"success": false, "message": "task plugin is still in use", "data": gin.H{"channels": channels, "in_flight_count": inFlight}})
return
}
}
_, err := model.DeleteTaskPluginVersion(key, version)
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
common.ApiErrorMsg(c, "override plugin version not found; factory plugins cannot be deleted")
return
}
common.ApiError(c, err)
return
}
if err = syncTaskPluginsOnceContext(c.Request.Context()); err != nil {
common.ApiError(c, err)
return
}
var data any
if plugin.Active {
// Deleting the active version changes which plugin, if any, serves key.
data = gin.H{"pricing_issues": taskPluginPricingIssues(key)}
}
common.ApiSuccess(c, data)
}
type taskPluginActivateRequest struct {
Version string `json:"version" binding:"required"`
}
func ActivateTaskPlugin(c *gin.Context) {
var request taskPluginActivateRequest
if err := c.ShouldBindJSON(&request); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
versions, err := model.ListTaskPluginVersions(c.Param("key"))
if err != nil {
common.ApiError(c, err)
return
}
var target *model.TaskPlugin
for i := range versions {
if versions[i].Version == request.Version {
target = &versions[i]
break
}
}
if target == nil {
common.ApiErrorMsg(c, "plugin version not found")
return
}
loaded, err := jsplugin.NewRegistry().Register(string(target.Source), jsplugin.Options{Key: target.Key, Version: target.Version})
if err != nil {
taskPluginCompileError(c, err)
return
}
// Activating a version also switches it on.
if err = jsplugin.PreflightRoutingConflict(jsplugin.DefaultRegistry.Generation(), loaded); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
if err = model.ActivateTaskPlugin(target.Key, target.Version); err != nil {
common.ApiError(c, err)
return
}
if err = syncTaskPluginsOnceContext(c.Request.Context()); err != nil {
common.ApiError(c, err)
return
}
common.ApiSuccess(c, gin.H{"pricing_issues": taskPluginPricingIssues(target.Key)})
}
type taskPluginStatusRequest struct {
Enabled *bool `json:"enabled" binding:"required"`
}
func SetTaskPluginStatus(c *gin.Context) {
var request taskPluginStatusRequest
if err := c.ShouldBindJSON(&request); err != nil || request.Enabled == nil {
common.ApiErrorMsg(c, "enabled is required")
return
}
key := c.Param("key")
disabledChannels := 0
unboundChannels := 0
if !*request.Enabled {
channels, inFlight, usageErr := model.GetTaskPluginUsage(key)
if usageErr != nil {
common.ApiError(c, usageErr)
return
}
cascade := c.Query("cascade") == "true"
force := c.Query("force") == "true"
if (len(channels) > 0 && !cascade) || (inFlight > 0 && !force) {
c.JSON(200, gin.H{"success": false, "message": "task plugin is still in use", "data": gin.H{"channels": channels, "in_flight_count": inFlight}})
return
}
if cascade {
for _, channel := range channels {
if channel.Type == constant.ChannelTypeNewAPI {
// A gateway channel also carries ordinary traffic, so the
// cascade only drops this plugin from its bindings.
unbound, unbindErr := model.UnbindTaskPlugin(channel.Id, key)
if unbindErr != nil {
common.ApiError(c, unbindErr)
return
}
if unbound {
unboundChannels++
}
continue
}
if model.UpdateChannelStatus(channel.Id, "", common.ChannelStatusManuallyDisabled, "task plugin disabled") {
disabledChannels++
}
}
if unboundChannels > 0 {
model.InitChannelCache()
}
}
}
active, lookupErr := model.GetTaskPluginVersion(key, "")
hasActiveOverride := lookupErr == nil
if lookupErr != nil && !errors.Is(lookupErr, gorm.ErrRecordNotFound) {
common.ApiError(c, lookupErr)
return
}
// Switching on serves the active version, or the factory built-in when the
// key has none.
if *request.Enabled && rejectTaskPluginRoutingConflict(c, key, active) {
return
}
response := gin.H{"plugin_enabled": *request.Enabled, "disabled_channels": disabledChannels, "unbound_channels": unboundChannels}
// Switching a key off must silence every layer that can serve it. The
// factory built-in goes into the disabled set even when an override row
// exists; otherwise the built-in would keep routing the same models (and
// blocking same-model uploads) right after the administrator disabled the
// plugin. Switching on reverses both layers.
if taskPluginHasFactory(key) {
keys := setting.GetTaskPluginDisabledFactoryKeys()
if *request.Enabled {
next := make([]string, 0, len(keys))
for _, item := range keys {
if item != key {
next = append(next, item)
}
}
keys = next
} else {
keys = append(append([]string{}, keys...), key)
}
if err := setting.SetTaskPluginDisabledFactoryKeysOption(keys); err != nil {
common.ApiError(c, err)
return
}
encoded, err := common.Marshal(setting.GetTaskPluginDisabledFactoryKeys())
if err != nil {
common.ApiError(c, err)
return
}
if err = model.UpdateOption(setting.TaskPluginDisabledFactoryKeysKey, string(encoded)); err != nil {
common.ApiError(c, err)
return
}
if !hasActiveOverride {
if *request.Enabled {
response["pricing_issues"] = taskPluginPricingIssues(key)
}
common.ApiSuccess(c, response)
return
}
}
if err := model.SetTaskPluginEnabled(key, *request.Enabled); err != nil {
common.ApiError(c, err)
return
}
if err := syncTaskPluginsOnceContext(c.Request.Context()); err != nil {
common.ApiError(c, err)
return
}
if *request.Enabled {
response["pricing_issues"] = taskPluginPricingIssues(key)
}
common.ApiSuccess(c, response)
}
func taskPluginHasFactory(key string) bool {
for _, meta := range jsplugin.DefaultRegistry.Snapshot().Factory {
if meta.Key == key {
return true
}
}
return false
}
// rejectTaskPluginRoutingConflict answers the request and returns true when
// the plugin about to serve key fails to compile or does not fit the live
// routing generation: row as the key's active version switched on, or the
// factory built-in when row is nil. A key without a built-in to fall back to
// has nothing to check. Keeping every enabled plugin conflict-free leaves the
// restart order nothing to decide.
func rejectTaskPluginRoutingConflict(c *gin.Context, key string, row *model.TaskPlugin) bool {
var candidate *jsplugin.LoadedPlugin
var err error
if row != nil {
candidate, err = jsplugin.CompilePlugin(string(row.Source), jsplugin.Options{Key: row.Key, Version: row.Version})
} else if taskPluginHasFactory(key) {
var source string
if source, err = plugins.Source(key); err == nil {
candidate, err = jsplugin.CompilePlugin(source, jsplugin.Options{Key: key})
}
}
if err != nil {
taskPluginCompileError(c, err)
return true
}
if candidate == nil {
return false
}
if err = jsplugin.PreflightRoutingConflict(jsplugin.DefaultRegistry.Generation(), candidate); err != nil {
common.ApiErrorMsg(c, err.Error())
return true
}
return false
}
// taskPluginPricingIssues lists, after the registry sync, the task model prices
// the plugin serving key cannot charge. The plugin change is already saved, so
// a failed read only drops the notice; the model pricing page still shows them.
func taskPluginPricingIssues(key string) []model.ModelPricingIssue {
issues, err := model.GetModelPricingIssues(key)
if err != nil {
common.SysError(common.LogText("read pricing issues of task plugin %s: %v", key, err))
return nil
}
return issues
}
func GetTaskPluginMarketplaceSources(c *gin.Context) {
common.ApiSuccess(c, setting.GetTaskPluginMarketplaceSources())
}
func UpdateTaskPluginMarketplaceSources(c *gin.Context) {
var sources []setting.TaskPluginMarketplaceSource
if err := c.ShouldBindJSON(&sources); err != nil {
common.ApiErrorMsg(c, err.Error())
return
}
if sources == nil {
sources = []setting.TaskPluginMarketplaceSource{}
}
for i := range sources {
name := strings.TrimSpace(sources[i].Name)
indexURL := strings.TrimSpace(sources[i].IndexURL)
if name == "" {
common.ApiErrorMsg(c, "marketplace source name is required")
return
}
parsed, err := url.Parse(indexURL)
if err != nil || !parsed.IsAbs() || parsed.Host == "" || (!strings.EqualFold(parsed.Scheme, "http") && !strings.EqualFold(parsed.Scheme, "https")) {
common.ApiErrorMsg(c, "marketplace source index_url must be an absolute http(s) URL")
return
}
sources[i].Name = name
sources[i].IndexURL = indexURL
}
encoded, err := common.Marshal(sources)
if err != nil {
common.ApiError(c, err)
return
}
if err = model.UpdateOption(setting.TaskPluginMarketplaceSourcesKey, string(encoded)); err != nil {
common.ApiError(c, err)
return
}
common.ApiSuccess(c, sources)
}
func GetTaskPluginOptions(c *gin.Context) {
snapshot := jsplugin.DefaultRegistry.Snapshot()
seen := make(map[string]bool)
options := make([]gin.H, 0, len(snapshot.Factory)+len(snapshot.Override))
for layer, metas := range [][]jsplugin.Meta{snapshot.Override, snapshot.Factory} {
for _, meta := range metas {
if seen[meta.Key] {
continue
}
// Disabled factory keys are omitted from bind options. The disabled
// set suppresses only the factory fallback; an enabled override for
// the same key is listed in the override pass and still appears.
if layer == 1 && setting.IsTaskPluginFactoryDisabled(meta.Key) {
continue
}
if _, ok := jsplugin.DefaultRegistry.Get(meta.Key); !ok {
continue
}
seen[meta.Key] = true
hasIcon := false
if layer == 0 {
if row, rowErr := model.GetTaskPluginVersion(meta.Key, ""); rowErr == nil {
hasIcon = row.HasIcon()
}
} else {
_, _, hasIcon = plugins.Icon(meta.Key)
}
options = append(options, gin.H{
"key": meta.Key,
"name": meta.Name,
"description": meta.Description,
"icon": meta.Icon,
"hasIcon": hasIcon,
"baseUrl": meta.BaseURL,
"sortPriority": meta.SortPriority,
"website": meta.Website,
"models": meta.Models,
"channelTypes": meta.ChannelTypes,
"upstreams": meta.Upstreams,
"usageSchema": meta.UsageSchema,
"usageProfiles": meta.UsageProfiles,
})
}
}
sort.Slice(options, func(i, j int) bool {
left, right := options[i]["sortPriority"].(int), options[j]["sortPriority"].(int)
if left != right {
return left > right
}
return options[i]["key"].(string) < options[j]["key"].(string)
})
common.ApiSuccess(c, options)
}
var taskPluginSyncState = struct {
sync.Mutex
hashes map[string]string
errors map[string]string
lastRebuild taskPluginRebuildOutcome
}{hashes: map[string]string{}, errors: map[string]string{}}
func syncTaskPluginsOnce() error {
return syncTaskPluginsOnceContext(context.Background())
}
func syncTaskPluginsOnceContext(ctx context.Context) error {
started := time.Now()
taskPluginSyncState.Lock()
defer taskPluginSyncState.Unlock()
databaseSnapshot, err := model.GetTaskPluginSyncSnapshot()
if err != nil {
syncErr := fmt.Errorf("sync task plugins: %w", err)
taskPluginSyncState.lastRebuild = taskPluginRebuildOutcome{
Status: "failed",
AttemptedAt: time.Now(),
Generation: jsplugin.DefaultRegistry.Generation().Number,
DatabaseRevision: taskPluginSyncState.lastRebuild.DatabaseRevision,
Error: syncErr.Error(),
}
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=failed stage=database_snapshot retained_generation=%d elapsed_ms=%d",
jsplugin.DefaultRegistry.Generation().Number,
time.Since(started).Milliseconds(),
)
return syncErr
}
databasePlugins := databaseSnapshot.Plugins
sort.Slice(databasePlugins, func(i, j int) bool { return databasePlugins[i].Key < databasePlugins[j].Key })
currentOverrides := jsplugin.DefaultRegistry.OverridePlugins()
generationBefore := jsplugin.DefaultRegistry.Generation().Number
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=start database_revision=%q generation=%d desired_plugins=%d current_overrides=%d",
databaseSnapshot.Revision,
generationBefore,
len(databasePlugins),
len(currentOverrides),
)
nextOverrides := make([]*jsplugin.LoadedPlugin, 0, len(databasePlugins))
nextHashes := make(map[string]string, len(databasePlugins))
seen := make(map[string]bool, len(databasePlugins))
for _, plugin := range databasePlugins {
seen[plugin.Key] = true
if current := currentOverrides[plugin.Key]; current != nil && taskPluginSyncState.hashes[plugin.Key] == plugin.SourceHash {
nextOverrides = append(nextOverrides, current)
nextHashes[plugin.Key] = plugin.SourceHash
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=plugin plugin=%q version=%q action=reuse",
plugin.Key,
plugin.Version,
)
continue
}
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=plugin plugin=%q version=%q action=compile_start",
plugin.Key,
plugin.Version,
)
var compiled *jsplugin.LoadedPlugin
source, compileErr := model.GetTaskPluginSource(plugin.Id)
if compileErr != nil {
compileErr = fmt.Errorf("load source: %w", compileErr)
} else {
compiled, compileErr = jsplugin.CompilePlugin(string(source), jsplugin.Options{Key: plugin.Key, Version: plugin.Version})
}
if compileErr != nil {
retainedIncumbent := false
if current := currentOverrides[plugin.Key]; current != nil {
nextOverrides = append(nextOverrides, current)
retainedIncumbent = true
if currentHash := taskPluginSyncState.hashes[plugin.Key]; currentHash != "" {
nextHashes[plugin.Key] = currentHash
}
}
taskPluginSyncState.errors[plugin.Key] = compileErr.Error()
common.SysError(common.LogText("compile task plugin %s@%s: %v", plugin.Key, plugin.Version, compileErr))
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=plugin plugin=%q version=%q action=compile_failed retained_incumbent=%t",
plugin.Key,
plugin.Version,
retainedIncumbent,
)
continue
}
nextOverrides = append(nextOverrides, compiled)
nextHashes[plugin.Key] = plugin.SourceHash
delete(taskPluginSyncState.errors, plugin.Key)
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=plugin plugin=%q version=%q action=compile_success",
plugin.Key,
plugin.Version,
)
}
if err = jsplugin.DefaultRegistry.ReplaceOverrides(nextOverrides); err != nil {
syncErr := fmt.Errorf("publish task plugin generation: %w", err)
taskPluginSyncState.lastRebuild = taskPluginRebuildOutcome{
Status: "failed",
AttemptedAt: time.Now(),
Generation: jsplugin.DefaultRegistry.Generation().Number,
DatabaseRevision: databaseSnapshot.Revision,
Error: syncErr.Error(),
}
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=failed stage=publish retained_generation=%d retained_generation_active=true database_revision=%q elapsed_ms=%d",
jsplugin.DefaultRegistry.Generation().Number,
databaseSnapshot.Revision,
time.Since(started).Milliseconds(),
)
return syncErr
}
taskPluginSyncState.hashes = nextHashes
for key := range taskPluginSyncState.errors {
if !seen[key] {
delete(taskPluginSyncState.errors, key)
}
}
pluginErrors := jsplugin.DefaultRegistry.RoutingErrors()
maps.Copy(pluginErrors, taskPluginSyncState.errors)
pluginErrorCount := len(pluginErrors)
status := "success"
if pluginErrorCount > 0 {
status = "partial"
}
taskPluginSyncState.lastRebuild = taskPluginRebuildOutcome{
Status: status,
AttemptedAt: time.Now(),
Generation: jsplugin.DefaultRegistry.Generation().Number,
DatabaseRevision: databaseSnapshot.Revision,
PluginErrorCount: pluginErrorCount,
}
logger.LogDebug(
ctx,
"task_plugin subsystem=sync event=complete database_revision=%q previous_generation=%d generation=%d status=%q active_overrides=%d plugin_errors=%d elapsed_ms=%d",
databaseSnapshot.Revision,
generationBefore,
jsplugin.DefaultRegistry.Generation().Number,
status,
len(jsplugin.DefaultRegistry.ActiveOverridePlugins()),
pluginErrorCount,
time.Since(started).Milliseconds(),
)
return nil
}
func SyncTaskPluginsOnce() {
if err := syncTaskPluginsOnce(); err != nil {
common.SysError(err.Error())
}
}
func SyncTaskPlugins() {
SyncTaskPluginsOnce()
for range time.NewTicker(30 * time.Second).C {
SyncTaskPluginsOnce()
}
}