Repository: langgenius/dify-plugin-daemon

Issue:
feat:.env.example is missing the DB_TYPE
**Self Checks**

To make sure we get to you in time, please check the following :)
- [x] I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify-plugin-daemon/issues), including closed ones.
- [x] I confirm that I am using English to submit this report (我已阅读并同意 [Language Policy](https://github.com/langgenius/dify/issues/1542)).
- [x] [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)
- [x] "Please do not modify this template :) and fill in all the required fields."


**Is your feature request related to a problem? Please describe.**
I find there is no DB_TYPE in .env.example.And need to consider some mysql-compatible db contidion.

**Describe the solution you'd like**
DB_type need to be added in .env.example.And for mysql-compatible database,just transfer DB_TYPE to mysql directly which is ok.
**Describe alternatives you've considered**
For mysql-compatible database,just transfer DB_TYPE to mysql directly which is ok such as oceanbase and seekdb.
**Additional context**
Add any other context or screenshots about the feature request here.
no.

Hints from issue comments:

Solution patch:
diff --git a/.env.example b/.env.example
index b1e9e8e..d248ef5 100644
--- a/.env.example
+++ b/.env.example
@@ -91,6 +91,7 @@ REDIS_SENTINEL_USERNAME=
 REDIS_SENTINEL_PASSWORD=
 REDIS_SENTINEL_SOCKET_TIMEOUT=0.1
 
+DB_TYPE=postgresql
 DB_USERNAME=postgres
 DB_PASSWORD=difyai123456
 DB_HOST=localhost
diff --git a/internal/core/control_panel/installer_local.go b/internal/core/control_panel/installer_local.go
index 71e0900..1243f14 100644
--- a/internal/core/control_panel/installer_local.go
+++ b/internal/core/control_panel/installer_local.go
@@ -6,7 +6,7 @@ import (
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
 )
 
-// InstallToLocalFromPkg installs a plugin to the local plugin runtime
+// InstallToLocal installs a plugin to the local plugin runtime
 // It's scope only for marking the plugin as `installed`,
 // you should call `LaunchLocalPlugin` to start it or it may launched by daemon
 // automatically
diff --git a/internal/core/control_panel/launcher_local.go b/internal/core/control_panel/launcher_local.go
index 18b9753..0aeee6c 100644
--- a/internal/core/control_panel/launcher_local.go
+++ b/internal/core/control_panel/launcher_local.go
@@ -26,6 +26,7 @@ func (c *ControlPanel) LaunchLocalPlugin(
 
 	// check if the plugin is already installed
 	if _, exists := c.localPluginRuntimes.Load(pluginUniqueIdentifier); exists {
+		c.localPluginInstallationLock.Unlock(pluginUniqueIdentifier.String())
 		return nil, nil, ErrorPluginAlreadyLaunched
 	}
 
diff --git a/internal/core/plugin_manager/installer.go b/internal/core/plugin_manager/installer.go
index 7115628..d45d69e 100644
--- a/internal/core/plugin_manager/installer.go
+++ b/internal/core/plugin_manager/installer.go
@@ -33,6 +33,24 @@ func (p *PluginManager) Install(
 	return p.installServerless(pluginUniqueIdentifier)
 }
 
+// SwitchServerlessEndpoint is required by enterprise dashboard.
+// one typical use case is rolling back to a previous version if the plugin is re-built,
+// while the dependencies of newer version break the functionality.
+func (p *PluginManager) SwitchServerlessEndpoint(
+	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
+	functionName string,
+	functionURL string,
+) error {
+	if p.config.Platform == app.PLATFORM_LOCAL {
+		return errors.New("switch serverless runtime is not supported on local platform")
+	}
+	err := p.updateServerlessRuntimeModel(pluginUniqueIdentifier, functionURL, functionName)
+	if err != nil {
+		return err
+	}
+	return p.clearServerlessRuntimeCache(pluginUniqueIdentifier)
+}
+
 // serverless runtime uses a strategy that firstly compile the plugin into a docker image
 // then execute it on a docker container(k8s pod / aws lambda)
 // however, it's mutable due to dependencies updates when using a version range to
diff --git a/internal/core/plugin_manager/manager.go b/internal/core/plugin_manager/manager.go
index d4ada73..e0858ee 100644
--- a/internal/core/plugin_manager/manager.go
+++ b/internal/core/plugin_manager/manager.go
@@ -11,6 +11,7 @@ import (
 	"github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation/calldify"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/media_transport"
 	serverless "github.com/langgenius/dify-plugin-daemon/internal/core/serverless_connector"
+	"github.com/langgenius/dify-plugin-daemon/internal/service/install_service"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
 	"github.com/langgenius/dify-plugin-daemon/pkg/plugin_packager/decoder"
@@ -83,8 +84,9 @@ func InitGlobalManager(oss oss.OSS, config *app.Config) *PluginManager {
 		config: config,
 	}
 
-	// mount logger to control panel
+	// mount control panel notifiers
 	manager.controlPanel.AddNotifier(&controlpanel.StandardLogger{})
+	manager.controlPanel.AddNotifier(&install_service.InstallListener{})
 
 	return manager
 }
diff --git a/internal/server/controllers/plugins.go b/internal/server/controllers/plugins.go
index f969e23..187a02d 100644
--- a/internal/server/controllers/plugins.go
+++ b/internal/server/controllers/plugins.go
@@ -10,7 +10,6 @@ import (
 	"github.com/langgenius/dify-plugin-daemon/internal/service"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/exception"
-	"github.com/langgenius/dify-plugin-daemon/pkg/entities/constants"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
 )
 
@@ -99,10 +98,6 @@ func UpgradePlugin(app *app.Config) gin.HandlerFunc {
 			Source                         string                                 `json:"source" validate:"required"`
 			Meta                           map[string]any                         `json:"meta" validate:"omitempty"`
 		}) {
-			if request.TenantID == constants.GlobalTenantId && !app.PluginAllowOrphans {
-				c.JSON(http.StatusOK, exception.BadRequestError(errors.New("orphan plugin is not allowed")).ToResponse())
-				return
-			}
 			if request.OriginalPluginUniqueIdentifier == request.NewPluginUniqueIdentifier {
 				c.JSON(http.StatusOK, exception.BadRequestError(errors.New("original and new plugin unique identifier are the same")).ToResponse())
 				return
@@ -136,10 +131,6 @@ func InstallPluginFromIdentifiers(app *app.Config) gin.HandlerFunc {
 			if request.Metas == nil {
 				request.Metas = []map[string]any{}
 			}
-			if request.TenantID == constants.GlobalTenantId && !app.PluginAllowOrphans {
-				c.JSON(http.StatusOK, exception.BadRequestError(errors.New("orphan plugin is not allowed")).ToResponse())
-				return
-			}
 
 			if len(request.Metas) != len(request.PluginUniqueIdentifiers) {
 				c.JSON(http.StatusOK, exception.BadRequestError(errors.New("the number of metas must be equal to the number of plugin unique identifiers")).ToResponse())
@@ -311,3 +302,20 @@ func ExtractPluginAsset(c *gin.Context) {
 		c.Data(http.StatusOK, "application/octet-stream", asset)
 	})
 }
+
+func SwitchServerlessEndpoint(c *gin.Context) {
+	BindRequest(c, func(request struct {
+		PluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier `json:"plugin_unique_identifier" validate:"required,plugin_unique_identifier"`
+		FunctionName           string                                 `json:"function_name" validate:"required"`
+		FunctionURL            string                                 `json:"function_url" validate:"required"`
+	}) {
+		c.JSON(
+			http.StatusOK,
+			service.SwitchServerlessEndpoint(
+				request.PluginUniqueIdentifier,
+				request.FunctionName,
+				request.FunctionURL,
+			),
+		)
+	})
+}
diff --git a/internal/server/http_server.go b/internal/server/http_server.go
index a62daf3..bf134cd 100644
--- a/internal/server/http_server.go
+++ b/internal/server/http_server.go
@@ -176,6 +176,7 @@ func (app *App) pluginManagementGroup(group *gin.RouterGroup, config *app.Config
 
 func (app *App) adminGroup(group *gin.RouterGroup, config *app.Config) {
 	group.POST("/plugin/serverless/reinstall", controllers.ReinstallPluginFromIdentifier(config))
+	group.POST("/plugin/serverless/switch-endpoint", controllers.SwitchServerlessEndpoint)
 }
 
 func (app *App) pluginAssetGroup(group *gin.RouterGroup) {
diff --git a/internal/server/middleware.go b/internal/server/middleware.go
index dbf43b4..3e78f2c 100644
--- a/internal/server/middleware.go
+++ b/internal/server/middleware.go
@@ -158,21 +158,15 @@ func (app *App) redirectPluginInvokeByPluginIdentifier(
 		}
 	}
 
-	for {
-		buf := make([]byte, 1024)
-		n, err := body.Read(buf)
-		if err != nil && err != io.EOF {
-			break
-		} else if err != nil {
-			ctx.Writer.Write(buf[:n])
-			ctx.Writer.Flush()
-			break
+	defer func(body io.ReadCloser) {
+		err := body.Close()
+		if err != nil {
+			log.Error("body close failed: %s", err.Error())
 		}
+	}(body)
 
-		if n > 0 {
-			ctx.Writer.Write(buf[:n])
-			ctx.Writer.Flush()
-		}
+	if _, err := io.Copy(ctx.Writer, body); err != nil {
+		log.Error("failed to write response body: %s", err.Error())
 	}
 }
 
diff --git a/internal/server/server.go b/internal/server/server.go
index ee3a876..2b8222f 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -117,6 +117,7 @@ func (app *App) Run(config *app.Config) {
 	tasks.SetupSignalHandler()
 	tasks.RegisterFinalizers(tasks.RecycleTasks)
 	tasks.RegisterFinalizers(cache.ReleaseAllLocks)
+	tasks.MonitorTimeoutTasks(app.cluster, config)
 
 	// start http server
 	app.server(config)
diff --git a/internal/service/install_plugin.go b/internal/service/install_plugin.go
index 671137d..2f34e13 100644
--- a/internal/service/install_plugin.go
+++ b/internal/service/install_plugin.go
@@ -14,7 +14,6 @@ import (
 	"github.com/langgenius/dify-plugin-daemon/internal/types/models"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/models/curd"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities"
-	"github.com/langgenius/dify-plugin-daemon/pkg/entities/constants"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/installation_entities"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
 	routinepkg "github.com/langgenius/dify-plugin-daemon/pkg/routine"
@@ -83,9 +82,7 @@ func InstallMultiplePluginsToTenant(
 		declarations = append(declarations, declaration)
 	}
 
-	// Always, EE edition needs a global reference to display installation progress
-	// add it if needed
-	tenants := joinGlobalTenantIfNeeded(config, tenantId)
+	tenants := []string{tenantId}
 
 	// all plugins are installed, no need to create tasks
 	// just add DB record and return
@@ -150,30 +147,6 @@ func ReinstallPluginFromIdentifier(
 	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
 ) {
 	baseSSEService(func() (*stream.Stream[installation_entities.PluginInstallResponse], error) {
-		pluginDeclaration, err := helper.CombinedGetPluginDeclaration(
-			pluginUniqueIdentifier,
-			plugin_entities.PLUGIN_RUNTIME_TYPE_SERVERLESS,
-		)
-		if err != nil {
-			return nil, errors.Join(err, errors.New("failed to get plugin declaration"))
-		}
-
-		plugin, err := db.GetOne[models.Plugin](
-			db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
-		)
-		if err != nil {
-			return nil, errors.Join(err, errors.New("failed to get plugin"))
-		}
-
-		statuses := buildTaskStatuses(
-			[]plugin_entities.PluginUniqueIdentifier{pluginUniqueIdentifier},
-			[]*plugin_entities.PluginDeclaration{pluginDeclaration},
-		)
-
-		taskRegistry, err := createInstallTasks([]string{constants.GlobalTenantId}, statuses)
-		if err != nil {
-			return nil, err
-		}
 
 		manager := plugin_manager.Manager()
 		if manager == nil {
@@ -189,7 +162,6 @@ func ReinstallPluginFromIdentifier(
 
 			reinstallStream, err := manager.Reinstall(pluginUniqueIdentifier)
 			if err != nil {
-				tasks.SetTaskStatusForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, models.InstallTaskStatusFailed, err.Error())
 				retStream.Write(installation_entities.PluginInstallResponse{
 					Event: installation_entities.PluginInstallEventError,
 					Data:  err.Error(),
@@ -199,30 +171,9 @@ func ReinstallPluginFromIdentifier(
 
 			err = reinstallStream.Process(func(resp installation_entities.PluginInstallResponse) {
 				retStream.Write(resp)
-				switch resp.Event {
-				case installation_entities.PluginInstallEventInfo:
-					tasks.SetTaskMessageForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, resp.Data)
-				case installation_entities.PluginInstallEventError:
-					tasks.SetTaskStatusForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, models.InstallTaskStatusFailed, resp.Data)
-				case installation_entities.PluginInstallEventDone:
-					_, _, installErr := curd.InstallPlugin(
-						constants.GlobalTenantId,
-						pluginUniqueIdentifier,
-						plugin.InstallType,
-						pluginDeclaration,
-						plugin.Source,
-						map[string]any{},
-					)
-					if installErr != nil && installErr != curd.ErrPluginAlreadyInstalled {
-						tasks.SetTaskStatusForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, models.InstallTaskStatusFailed, installErr.Error())
-						return
-					}
-					tasks.SetTaskStatusForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, models.InstallTaskStatusSuccess, "Reinstalled")
-				}
 			})
 
 			if err != nil {
-				tasks.SetTaskStatusForOnePlugin(taskRegistry.IDs(), pluginUniqueIdentifier, models.InstallTaskStatusFailed, err.Error())
 				retStream.Write(installation_entities.PluginInstallResponse{
 					Event: installation_entities.PluginInstallEventError,
 					Data:  err.Error(),
@@ -309,7 +260,7 @@ func UpgradePlugin(
 	}
 
 	// construct tenant jobs
-	tenants := joinGlobalTenantIfNeeded(config, tenantId)
+	tenants := []string{tenantId}
 
 	job := tasks.PluginUpgradeJob{
 		NewIdentifier:       newPluginUniqueIdentifier,
diff --git a/internal/service/install_plugin_enterprise.go b/internal/service/install_plugin_enterprise.go
deleted file mode 100644
index 7beb0af..0000000
--- a/internal/service/install_plugin_enterprise.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package service
-
-// NOTE: ENTERPRISE ONLY
-
-import (
-	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
-	"github.com/langgenius/dify-plugin-daemon/pkg/entities/constants"
-)
-
-func joinGlobalTenantIfNeeded(config *app.Config, tenantId string) []string {
-	tenants := []string{tenantId}
-	if tenantId != constants.GlobalTenantId && config.PluginAllowOrphans {
-		tenants = append(tenants, constants.GlobalTenantId)
-	}
-	return tenants
-}
diff --git a/internal/service/install_service/controlpanel.go b/internal/service/install_service/controlpanel.go
index ce5a28c..32b0938 100644
--- a/internal/service/install_service/controlpanel.go
+++ b/internal/service/install_service/controlpanel.go
@@ -9,6 +9,30 @@ import (
 
 type InstallListener struct{}
 
+func (l *InstallListener) OnLocalRuntimeStarting(pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier) {
+}
+
+func (l *InstallListener) OnLocalRuntimeReady(runtime *local_runtime.LocalPluginRuntime) {
+}
+
+func (l *InstallListener) OnLocalRuntimeStartFailed(
+	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
+	err error,
+) {
+}
+
+func (l *InstallListener) OnLocalRuntimeStop(runtime *local_runtime.LocalPluginRuntime) {
+}
+
+func (l *InstallListener) OnLocalRuntimeStopped(runtime *local_runtime.LocalPluginRuntime) {
+}
+
+func (l *InstallListener) OnLocalRuntimeScaleUp(runtime *local_runtime.LocalPluginRuntime, instanceNums int32) {
+}
+
+func (l *InstallListener) OnLocalRuntimeScaleDown(runtime *local_runtime.LocalPluginRuntime, instanceNums int32) {
+}
+
 func (l *InstallListener) OnDebuggingRuntimeConnected(runtime *debugging_runtime.RemotePluginRuntime) {
 	_, installation, err := InstallPlugin(
 		runtime.TenantId(),
@@ -41,11 +65,3 @@ func (l *InstallListener) OnDebuggingRuntimeDisconnected(runtime *debugging_runt
 		log.Error("uninstall debugging plugin failed, error: %v", err)
 	}
 }
-
-func (l *InstallListener) OnLocalRuntimeReady(runtime *local_runtime.LocalPluginRuntime) {
-
-}
-
-func (l *InstallListener) OnLocalRuntimeStartFailed(runtime *local_runtime.LocalPluginRuntime) {
-
-}
diff --git a/internal/service/manage_plugin.go b/internal/service/manage_plugin.go
index ee59902..5933950 100644
--- a/internal/service/manage_plugin.go
+++ b/internal/service/manage_plugin.go
@@ -4,6 +4,7 @@ import (
 	"errors"
 	"time"
 
+	"github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager"
 	"github.com/langgenius/dify-plugin-daemon/internal/db"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/exception"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/models"
@@ -573,3 +574,15 @@ func GetDatasource(tenant_id string, plugin_id string, provider string) *entitie
 		Declaration:            declaration.Datasource,
 	})
 }
+
+func SwitchServerlessEndpoint(
+	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
+	functionName string,
+	functionURL string) *entities.Response {
+	manager := plugin_manager.Manager()
+	err := manager.SwitchServerlessEndpoint(pluginUniqueIdentifier, functionName, functionURL)
+	if err != nil {
+		return exception.InternalServerError(err).ToResponse()
+	}
+	return entities.NewSuccessResponse(true)
+}
diff --git a/internal/tasks/recycle.go b/internal/tasks/recycle.go
index e991168..71b422b 100644
--- a/internal/tasks/recycle.go
+++ b/internal/tasks/recycle.go
@@ -2,8 +2,11 @@ package tasks
 
 import (
 	"errors"
+	"time"
 
+	"github.com/langgenius/dify-plugin-daemon/internal/cluster"
 	"github.com/langgenius/dify-plugin-daemon/internal/db"
+	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/models"
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/log"
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/mapping"
@@ -60,3 +63,65 @@ func RecycleTasks() error {
 	})
 	return errors.Join(errs...)
 }
+
+func markTasksAsTimeout(tasks []*models.InstallTask) {
+	if len(tasks) == 0 {
+		return
+	}
+	for _, task := range tasks {
+		task.Status = models.InstallTaskStatusFailed
+		for i := range task.Plugins {
+			plugin := &task.Plugins[i]
+			plugin.Status = models.InstallTaskStatusFailed
+			plugin.Message = "Task timed out but not properly terminated"
+		}
+	}
+	err := db.Update(tasks)
+	if err != nil {
+		log.Error("failed to update tasks: %v", err)
+	}
+}
+
+// Just in case some tasks may stuck for some reason that we don't know.
+func MonitorTimeoutTasks(cluster *cluster.Cluster, config *app.Config) {
+	go func() {
+
+		var timeout time.Duration
+		if config.Platform == app.PLATFORM_SERVERLESS {
+			timeout = time.Duration(config.DifyPluginServerlessConnectorLaunchTimeout) * time.Second
+		} else {
+			timeout = time.Duration(config.PythonEnvInitTimeout) * time.Second
+		}
+		// add some tolerance to timeout to avoid race condition
+		timeout = timeout + time.Minute
+
+		ticker := time.NewTicker(time.Minute)
+		defer ticker.Stop()
+		for range ticker.C {
+			if !cluster.IsMaster() {
+				continue
+			}
+			tasksToProcess := []*models.InstallTask{}
+			// get all tasks that are pending or running
+			tasks, err := db.GetAll[models.InstallTask](
+				db.InArray("status", []any{
+					string(models.InstallTaskStatusPending),
+					string(models.InstallTaskStatusRunning),
+				}),
+			)
+			if err != nil {
+				log.Error("failed to get all tasks: %v", err)
+				continue
+			}
+			for i := range tasks {
+				task := &tasks[i]
+				if time.Since(task.CreatedAt) > timeout {
+					log.Info("task %s timed out", task.ID)
+					tasksToProcess = append(tasksToProcess, task)
+				}
+			}
+			markTasksAsTimeout(tasksToProcess)
+		}
+	}()
+
+}
diff --git a/internal/types/app/config.go b/internal/types/app/config.go
index e5c17f7..1808c1e 100644
--- a/internal/types/app/config.go
+++ b/internal/types/app/config.go
@@ -10,6 +10,8 @@ import (
 const (
 	DB_TYPE_POSTGRESQL = "postgresql"
 	DB_TYPE_MYSQL      = "mysql"
+	DB_TYPE_OCEANBASE = "oceanbase"
+	DB_TYPE_SEEKDB    = "seekdb"
 )
 
 type Config struct {
@@ -97,10 +99,6 @@ type Config struct {
 	// local launching max concurrent
 	PluginLocalLaunchingConcurrent int `envconfig:"PLUGIN_LOCAL_LAUNCHING_CONCURRENT" validate:"required"`
 
-	// add a global reference to plugins to prevent them from being garbage collected
-	// not allowed for local mode
-	PluginAllowOrphans bool `envconfig:"PLUGIN_ALLOW_ORPHANS" default:"false"`
-
 	// platform like local or aws lambda
 	Platform PlatformType `envconfig:"PLATFORM" validate:"required"`
 
diff --git a/internal/types/app/default.go b/internal/types/app/default.go
index dffe0fa..48f056a 100644
--- a/internal/types/app/default.go
+++ b/internal/types/app/default.go
@@ -6,6 +6,10 @@ import (
 )
 
 func (config *Config) SetDefault() {
+	switch config.DBType {
+	case DB_TYPE_OCEANBASE, DB_TYPE_SEEKDB:
+		config.DBType = DB_TYPE_MYSQL
+	}
 	setDefaultInt(&config.ServerPort, 5002)
 	setDefaultInt(&config.RoutinePoolSize, 10000)
 	setDefaultInt(&config.LifetimeCollectionGCInterval, 60)
diff --git a/internal/types/models/curd/atomic.go b/internal/types/models/curd/atomic.go
index 89b22c9..d002f9e 100644
--- a/internal/types/models/curd/atomic.go
+++ b/internal/types/models/curd/atomic.go
@@ -5,37 +5,11 @@ import (
 
 	"github.com/langgenius/dify-plugin-daemon/internal/db"
 	"github.com/langgenius/dify-plugin-daemon/internal/types/models"
-	"github.com/langgenius/dify-plugin-daemon/pkg/entities/constants"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/manifest_entities"
 	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
 	"gorm.io/gorm"
 )
 
-func EnsureGlobalReference(
-	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
-	tenantId string,
-	installType plugin_entities.PluginRuntimeType,
-	declaration *plugin_entities.PluginDeclaration,
-	source string,
-	meta map[string]any,
-) error {
-	if tenantId == constants.GlobalTenantId {
-		return nil
-	}
-	_, _, err := InstallPlugin(
-		constants.GlobalTenantId,
-		pluginUniqueIdentifier,
-		installType,
-		declaration,
-		source,
-		meta,
-	)
-	if err != nil && err != ErrPluginAlreadyInstalled {
-		return err
-	}
-	return nil
-}
-
 // Create plugin for a tenant, create plugin if it has never been created before
 // and install it to the tenant, return the plugin and the installation
 // if the plugin has been created before, return the plugin which has been created before
diff --git a/pkg/entities/constants/identity.go b/pkg/entities/constants/identity.go
index 629c1d7..1255fc8 100644
--- a/pkg/entities/constants/identity.go
+++ b/pkg/entities/constants/identity.go
@@ -1,7 +1 @@
 package constants
-
-const (
-	// GlobalTenantId not visible to end users.
-	// Used to prevent plugins from being garbage collected when uninstalled from all user workspaces.
-	GlobalTenantId = "00000000-0000-0000-0000-000000000000"
-)
diff --git a/pkg/entities/plugin_entities/tool_declaration.go b/pkg/entities/plugin_entities/tool_declaration.go
index 2c5d867..41891b3 100644
--- a/pkg/entities/plugin_entities/tool_declaration.go
+++ b/pkg/entities/plugin_entities/tool_declaration.go
@@ -132,9 +132,10 @@ type ToolParameter struct {
 	Required         bool                   `json:"required" yaml:"required"`
 	AutoGenerate     *ParameterAutoGenerate `json:"auto_generate" yaml:"auto_generate" validate:"omitempty"`
 	Template         *ParameterTemplate     `json:"template" yaml:"template" validate:"omitempty"`
-	Default          any                    `json:"default" yaml:"default" validate:"omitempty,is_basic_type"`
+	Default          any                    `json:"default" yaml:"default" validate:"omitempty"`
 	Min              *float64               `json:"min" yaml:"min" validate:"omitempty"`
 	Max              *float64               `json:"max" yaml:"max" validate:"omitempty"`
+	Multiple         bool                   `json:"multiple" yaml:"multiple" validate:"omitempty"`
 	Precision        *int                   `json:"precision" yaml:"precision" validate:"omitempty"`
 	Options          []ParameterOption      `json:"options" yaml:"options" validate:"omitempty,dive"`
 }
@@ -179,7 +180,7 @@ type ToolProviderIdentity struct {
 	Name        string                        `json:"name" validate:"required,tool_provider_identity_name"`
 	Description I18nObject                    `json:"description"`
 	Icon        string                        `json:"icon" validate:"required"`
-	IconDark    string                        `json:"icon_dark" validate:"omitempty"`
+	IconDark    string                        `json:"icon_dark" yaml:"icon_dark" validate:"omitempty"`
 	Label       I18nObject                    `json:"label" validate:"required"`
 	Tags        []manifest_entities.PluginTag `json:"tags" validate:"omitempty,dive,plugin_tag"`
 }
diff --git a/pkg/entities/plugin_entities/trigger_declaration.go b/pkg/entities/plugin_entities/trigger_declaration.go
index f338157..6982e6c 100644
--- a/pkg/entities/plugin_entities/trigger_declaration.go
+++ b/pkg/entities/plugin_entities/trigger_declaration.go
@@ -89,7 +89,7 @@ type TriggerProviderIdentity struct {
 	Name        string                        `json:"name" validate:"required,tool_provider_identity_name"`
 	Description I18nObject                    `json:"description"`
 	Icon        string                        `json:"icon" validate:"required"`
-	IconDark    string                        `json:"icon_dark" validate:"omitempty"`
+	IconDark    string                        `json:"icon_dark" yaml:"icon_dark" validate:"omitempty"`
 	Label       I18nObject                    `json:"label" validate:"required"`
 	Tags        []manifest_entities.PluginTag `json:"tags" validate:"omitempty,dive,plugin_tag"`
 }
diff --git a/pkg/utils/parser/chunking.go b/pkg/utils/parser/chunking.go
index 691b60c..b56d84b 100644
--- a/pkg/utils/parser/chunking.go
+++ b/pkg/utils/parser/chunking.go
@@ -24,7 +24,7 @@ func LineBasedChunking(reader io.Reader, maxChunkSize int, processor func([]byte
 			return err
 		}
 	}
-	return nil
+	return scanner.Err()
 }
 
 // We uses following format:



Test patch:
diff --git a/internal/cluster/redirect_test.go b/internal/cluster/redirect_test.go
index bf4b48a..cc44d02 100644
--- a/internal/cluster/redirect_test.go
+++ b/internal/cluster/redirect_test.go
@@ -106,6 +106,7 @@ func TestRedirectTraffic(t *testing.T) {
 						c.Header(k, vv)
 					}
 				}
+				defer reader.Close()
 				io.Copy(c.Writer, reader)
 			} else {
 				c.String(http.StatusOK, "ok")