dify-plugin-daemon/internal/service/install_plugin.go

package service

import (
	"errors"
	"fmt"
	"time"

	"github.com/gin-gonic/gin"
	"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/app"
	"github.com/langgenius/dify-plugin-daemon/internal/types/exception"
	"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/internal/utils/cache"
	"github.com/langgenius/dify-plugin-daemon/internal/utils/cache/helper"
	"github.com/langgenius/dify-plugin-daemon/internal/utils/log"
	"github.com/langgenius/dify-plugin-daemon/internal/utils/routine"
	"github.com/langgenius/dify-plugin-daemon/internal/utils/stream"
	"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/plugin_entities"
	"github.com/langgenius/dify-plugin-daemon/pkg/plugin_packager/decoder"
	"gorm.io/gorm"
)

type InstallPluginResponse struct {
	AllInstalled bool   `json:"all_installed"`
	TaskID       string `json:"task_id"`
}

type InstallPluginOnDoneHandler func(
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
	declaration *plugin_entities.PluginDeclaration,
	meta map[string]any,
) error

type InstallPluginOnMessageHandler func(
	message plugin_manager.PluginInstallResponse,
)

func doInstallPluginRuntime(
	runtimeType plugin_entities.PluginRuntimeType,
	manager *plugin_manager.PluginManager,
	config *app.Config,
	tenant_id string,
	source string,
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
	meta map[string]any,
	task *models.InstallTask,
	declaration *plugin_entities.PluginDeclaration,
	reinstall bool,
	onMessage InstallPluginOnMessageHandler,
	onDone InstallPluginOnDoneHandler,
) {
	var err error
	updateTaskStatus := func(modifier func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus)) {

		if err := db.WithTransaction(func(tx *gorm.DB) error {
			task, err := db.GetOne[models.InstallTask](
				db.WithTransactionContext(tx),
				db.Equal("id", task.ID),
				db.WLock(), // write lock, multiple tasks can't update the same task
			)

			if err == db.ErrDatabaseNotFound {
				return nil
			}

			if err != nil {
				return err
			}

			taskPointer := &task
			var pluginStatus *models.InstallTaskPluginStatus
			for i := range task.Plugins {
				if task.Plugins[i].PluginUniqueIdentifier == pluginUniqueIdentifier {
					pluginStatus = &task.Plugins[i]
					break
				}
			}

			if pluginStatus == nil {
				return nil
			}

			modifier(taskPointer, pluginStatus)

			successes := 0
			for _, plugin := range taskPointer.Plugins {
				if plugin.Status == models.InstallTaskStatusSuccess {
					successes++
				}
			}

			if successes == len(taskPointer.Plugins) {
				// update status
				taskPointer.Status = models.InstallTaskStatusSuccess
				// delete the task after 120 seconds without transaction
				time.AfterFunc(120*time.Second, func() {
					db.Delete(taskPointer)
				})
			}
			return db.Update(taskPointer, tx)
		}); err != nil {
			log.Error("failed to update install task status %s", err.Error())
		}
	}

	updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
		plugin.Status = models.InstallTaskStatusRunning
		plugin.Message = "Installing"
	})

	var stream *stream.Stream[plugin_manager.PluginInstallResponse]
	if config.Platform == app.PLATFORM_SERVERLESS {
		var zipDecoder *decoder.ZipPluginDecoder
		var pkgFile []byte

		pkgFile, err = manager.GetPackage(pluginUniqueIdentifier)
		if err != nil {
			updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
				task.Status = models.InstallTaskStatusFailed
				plugin.Status = models.InstallTaskStatusFailed
				plugin.Message = "Failed to read plugin package"
				onMessage(plugin_manager.PluginInstallResponse{
					Event: plugin_manager.PluginInstallEventError,
					Data:  plugin.Message,
				})
			})
			return
		}

		zipDecoder, err = decoder.NewZipPluginDecoderWithThirdPartySignatureVerificationConfig(
			pkgFile,
			&decoder.ThirdPartySignatureVerificationConfig{
				Enabled:        config.ThirdPartySignatureVerificationEnabled,
				PublicKeyPaths: config.ThirdPartySignatureVerificationPublicKeys,
			},
		)
		if err != nil {
			updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
				task.Status = models.InstallTaskStatusFailed
				plugin.Status = models.InstallTaskStatusFailed
				plugin.Message = err.Error()
				onMessage(plugin_manager.PluginInstallResponse{
					Event: plugin_manager.PluginInstallEventError,
					Data:  plugin.Message,
				})
			})
			return
		}
		if reinstall {
			stream, err = manager.ReinstallToServerlessFromPkg(pkgFile, zipDecoder)
		} else {
			stream, err = manager.InstallToServerlessFromPkg(pkgFile, zipDecoder, source, meta)
		}
	} else if config.Platform == app.PLATFORM_LOCAL {
		if reinstall {
			log.Warn("reinstall is not supported on local platform, will do install")
		}
		stream, err = manager.InstallToLocal(pluginUniqueIdentifier, source, meta)
	} else {
		updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
			task.Status = models.InstallTaskStatusFailed
			plugin.Status = models.InstallTaskStatusFailed
			plugin.Message = "Unsupported platform"
			onMessage(plugin_manager.PluginInstallResponse{
				Event: plugin_manager.PluginInstallEventError,
				Data:  plugin.Message,
			})
		})
		return
	}

	if err != nil {
		updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
			task.Status = models.InstallTaskStatusFailed
			plugin.Status = models.InstallTaskStatusFailed
			plugin.Message = err.Error()
			onMessage(plugin_manager.PluginInstallResponse{
				Event: plugin_manager.PluginInstallEventError,
				Data:  plugin.Message,
			})
		})
		return
	}

	for stream.Next() {
		message, err := stream.Read()
		if err != nil {
			updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
				task.Status = models.InstallTaskStatusFailed
				plugin.Status = models.InstallTaskStatusFailed
				plugin.Message = err.Error()
			})
			return
		}
		onMessage(message)
		if message.Event == plugin_manager.PluginInstallEventError {
			updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
				task.Status = models.InstallTaskStatusFailed
				plugin.Status = models.InstallTaskStatusFailed
				plugin.Message = message.Data
			})
			return
		}

		if message.Event == plugin_manager.PluginInstallEventDone {
			if err := curd.EnsureGlobalReferenceIfRequired(pluginUniqueIdentifier, tenant_id, runtimeType, declaration, source, meta); err != nil {
				updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
					task.Status = models.InstallTaskStatusFailed
					plugin.Status = models.InstallTaskStatusFailed
					plugin.Message = err.Error()
				})
				return
			}
			if err := onDone(pluginUniqueIdentifier, declaration, meta); err != nil {
				updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
					task.Status = models.InstallTaskStatusFailed
					plugin.Status = models.InstallTaskStatusFailed
					plugin.Message = "Failed to create plugin, perhaps it's already installed"
				})
				return
			}
		}
	}

	updateTaskStatus(func(task *models.InstallTask, plugin *models.InstallTaskPluginStatus) {
		plugin.Status = models.InstallTaskStatusSuccess
		plugin.Message = "Installed"
		task.CompletedPlugins++

		// check if all plugins are installed
		if task.CompletedPlugins == task.TotalPlugins {
			task.Status = models.InstallTaskStatusSuccess
		}
	})
}

func InstallPluginRuntimeToTenant(
	config *app.Config,
	tenant_id string,
	plugin_unique_identifiers []plugin_entities.PluginUniqueIdentifier,
	source string,
	metas []map[string]any,
	onDone InstallPluginOnDoneHandler, // since installing plugin is a async task, we need to call it asynchronously
) (*InstallPluginResponse, error) {
	response := &InstallPluginResponse{}
	pluginsWaitForInstallation := []plugin_entities.PluginUniqueIdentifier{}

	runtimeType := plugin_entities.PluginRuntimeType("")
	if config.Platform == app.PLATFORM_SERVERLESS {
		runtimeType = plugin_entities.PLUGIN_RUNTIME_TYPE_SERVERLESS
	} else if config.Platform == app.PLATFORM_LOCAL {
		runtimeType = plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL
	} else {
		return nil, fmt.Errorf("unsupported platform: %s", config.Platform)
	}

	task := &models.InstallTask{
		Status:           models.InstallTaskStatusRunning,
		TenantID:         tenant_id,
		TotalPlugins:     len(plugin_unique_identifiers),
		CompletedPlugins: 0,
		Plugins:          []models.InstallTaskPluginStatus{},
	}

	for i, pluginUniqueIdentifier := range plugin_unique_identifiers {
		// fetch plugin declaration first, before installing, we need to ensure pkg is uploaded
		pluginDeclaration, err := helper.CombinedGetPluginDeclaration(
			pluginUniqueIdentifier,
			runtimeType,
		)
		if err != nil {
			return nil, err
		}

		// check if plugin is already installed
		_, err = db.GetOne[models.Plugin](
			db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
		)

		task.Plugins = append(task.Plugins, models.InstallTaskPluginStatus{
			PluginUniqueIdentifier: pluginUniqueIdentifier,
			PluginID:               pluginUniqueIdentifier.PluginID(),
			Status:                 models.InstallTaskStatusPending,
			Icon:                   pluginDeclaration.Icon,
			IconDark:               pluginDeclaration.IconDark,
			Labels:                 pluginDeclaration.Label,
			Message:                "",
		})

		if err == nil {
			if err := curd.EnsureGlobalReferenceIfRequired(pluginUniqueIdentifier, tenant_id, runtimeType, pluginDeclaration, source, metas[i]); err != nil {
				return nil, err
			}
			if err := onDone(pluginUniqueIdentifier, pluginDeclaration, metas[i]); err != nil {
				return nil, errors.Join(err, errors.New("failed on plugin installation"))
			} else {
				task.CompletedPlugins++
				task.Plugins[i].Status = models.InstallTaskStatusSuccess
				task.Plugins[i].Message = "Installed"
			}

			continue
		}

		if err != db.ErrDatabaseNotFound {
			return nil, err
		}

		pluginsWaitForInstallation = append(pluginsWaitForInstallation, pluginUniqueIdentifier)
	}

	if len(pluginsWaitForInstallation) == 0 {
		response.AllInstalled = true
		response.TaskID = ""
		return response, nil
	}

	err := db.Create(task)
	if err != nil {
		return nil, err
	}

	response.TaskID = task.ID
	manager := plugin_manager.Manager()

	tasks := []func(){}
	for i, pluginUniqueIdentifier := range pluginsWaitForInstallation {
		declaration, err := helper.CombinedGetPluginDeclaration(
			pluginUniqueIdentifier,
			runtimeType,
		)
		if err != nil {
			return nil, err
		}

		i := i
		tasks = append(tasks, func() {
			doInstallPluginRuntime(
				runtimeType,
				manager,
				config,
				tenant_id,
				source,
				pluginUniqueIdentifier,
				metas[i],
				task,
				declaration,
				false,
				func(message plugin_manager.PluginInstallResponse) {},
				onDone)
		})
	}

	// submit async tasks
	routine.WithMaxRoutine(5, tasks)

	return response, nil
}

func InstallPluginFromIdentifiers(
	config *app.Config,
	tenant_id string,
	plugin_unique_identifiers []plugin_entities.PluginUniqueIdentifier,
	source string,
	metas []map[string]any,
) *entities.Response {
	response, err := InstallPluginRuntimeToTenant(
		config,
		tenant_id,
		plugin_unique_identifiers,
		source,
		metas,
		func(
			pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
			declaration *plugin_entities.PluginDeclaration,
			meta map[string]any,
		) error {
			runtimeType := plugin_entities.PluginRuntimeType("")

			switch config.Platform {
			case app.PLATFORM_SERVERLESS:
				runtimeType = plugin_entities.PLUGIN_RUNTIME_TYPE_SERVERLESS
			case app.PLATFORM_LOCAL:
				runtimeType = plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL
			default:
				return fmt.Errorf("unsupported platform: %s", config.Platform)
			}
			_, _, err := curd.InstallPlugin(tenant_id, pluginUniqueIdentifier, runtimeType, declaration, source, meta)
			return err
		},
	)
	if err != nil {
		if errors.Is(err, curd.ErrPluginAlreadyInstalled) {
			return exception.BadRequestError(err).ToResponse()
		}
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(response)
}

/*
 * Reinstall a plugin from a given identifier, no tenant_id is needed
 */
func ReinstallPluginFromIdentifier(
	ctx *gin.Context,
	config *app.Config,
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
) {
	baseSSEService(func() (*stream.Stream[plugin_manager.PluginInstallResponse], error) {
		pluginDeclaration, err := helper.CombinedGetPluginDeclaration(
			pluginUniqueIdentifier,
			plugin_entities.PLUGIN_RUNTIME_TYPE_SERVERLESS,
		)
		if err != nil {
			return nil, err
		}

		plugin, err := db.GetOne[models.Plugin](
			db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
		)
		if err != nil {
			return nil, err
		}

		retStream := stream.NewStream[plugin_manager.PluginInstallResponse](128)

		task := &models.InstallTask{
			Status:           models.InstallTaskStatusRunning,
			TenantID:         constants.GlobalTenantId,
			TotalPlugins:     1,
			CompletedPlugins: 0,
			Plugins:          []models.InstallTaskPluginStatus{},
		}
		task.Plugins = append(task.Plugins, models.InstallTaskPluginStatus{
			PluginUniqueIdentifier: pluginUniqueIdentifier,
			PluginID:               pluginUniqueIdentifier.PluginID(),
			Status:                 models.InstallTaskStatusPending,
			Icon:                   pluginDeclaration.Icon,
			IconDark:               pluginDeclaration.IconDark,
			Labels:                 pluginDeclaration.Label,
			Message:                "",
		})

		err = db.Create(task)
		if err != nil {
			return nil, err
		}

		f := func() {
			doInstallPluginRuntime(
				plugin_entities.PLUGIN_RUNTIME_TYPE_SERVERLESS,
				plugin_manager.Manager(),
				config,
				constants.GlobalTenantId,
				plugin.Source,
				pluginUniqueIdentifier,
				map[string]any{},
				task,
				pluginDeclaration,
				true,
				func(message plugin_manager.PluginInstallResponse) {
					retStream.Write(message)
				},
				func(pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier, declaration *plugin_entities.PluginDeclaration, meta map[string]any) error {
					retStream.Close()
					return nil
				})
		}
		routine.Submit(nil, f)
		return retStream, nil
	}, ctx, 1800)
}

/*
 * Decode a plugin from a given identifier, no tenant_id is needed
 * When upload local plugin inside Dify, the second step need to ensure that the plugin is valid
 * So we need to provide a way to decode the plugin and verify the signature
 */
func DecodePluginFromIdentifier(
	config *app.Config,
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
) *entities.Response {
	// get plugin package and decode again
	manager := plugin_manager.Manager()
	pkgFile, err := manager.GetPackage(pluginUniqueIdentifier)
	if err != nil {
		return exception.BadRequestError(err).ToResponse()
	}

	zipDecoder, err := decoder.NewZipPluginDecoderWithThirdPartySignatureVerificationConfig(
		pkgFile,
		&decoder.ThirdPartySignatureVerificationConfig{
			Enabled:        config.ThirdPartySignatureVerificationEnabled,
			PublicKeyPaths: config.ThirdPartySignatureVerificationPublicKeys,
		},
	)
	if err != nil {
		return exception.BadRequestError(err).ToResponse()
	}

	verification, _ := zipDecoder.Verification()
	if verification == nil && zipDecoder.Verified() {
		verification = decoder.DefaultVerification()
	}

	declaration, err := zipDecoder.Manifest()
	if err != nil {
		return exception.BadRequestError(err).ToResponse()
	}

	return entities.NewSuccessResponse(map[string]any{
		"unique_identifier": pluginUniqueIdentifier,
		"manifest":          declaration,
		"verification":      verification,
	})
}

func UpgradePlugin(
	config *app.Config,
	tenant_id string,
	source string,
	meta map[string]any,
	original_plugin_unique_identifier plugin_entities.PluginUniqueIdentifier,
	new_plugin_unique_identifier plugin_entities.PluginUniqueIdentifier,
) *entities.Response {
	if original_plugin_unique_identifier == new_plugin_unique_identifier {
		return exception.BadRequestError(errors.New("original and new plugin unique identifier are the same")).ToResponse()
	}

	if original_plugin_unique_identifier.PluginID() != new_plugin_unique_identifier.PluginID() {
		return exception.BadRequestError(errors.New("original and new plugin id are different")).ToResponse()
	}

	// uninstall the original plugin
	installation, err := db.GetOne[models.PluginInstallation](
		db.Equal("tenant_id", tenant_id),
		db.Equal("plugin_unique_identifier", original_plugin_unique_identifier.String()),
		db.Equal("source", source),
	)

	if err == db.ErrDatabaseNotFound {
		return exception.NotFoundError(errors.New("plugin installation not found for this tenant")).ToResponse()
	}

	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	// install the new plugin runtime
	response, err := InstallPluginRuntimeToTenant(
		config,
		tenant_id,
		[]plugin_entities.PluginUniqueIdentifier{new_plugin_unique_identifier},
		source,
		[]map[string]any{meta},
		func(
			pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
			declaration *plugin_entities.PluginDeclaration,
			meta map[string]any,
		) error {
			originalDeclaration, err := helper.CombinedGetPluginDeclaration(
				original_plugin_unique_identifier,
				plugin_entities.PluginRuntimeType(installation.RuntimeType),
			)
			if err != nil {
				return err
			}

			newDeclaration, err := helper.CombinedGetPluginDeclaration(
				new_plugin_unique_identifier,
				plugin_entities.PluginRuntimeType(installation.RuntimeType),
			)
			if err != nil {
				return err
			}
			// uninstall the original plugin
			upgradeResponse, err := curd.UpgradePlugin(
				tenant_id,
				original_plugin_unique_identifier,
				new_plugin_unique_identifier,
				originalDeclaration,
				newDeclaration,
				plugin_entities.PluginRuntimeType(installation.RuntimeType),
				source,
				meta,
			)

			if err != nil {
				return err
			}

			// invalidate plugin installation cache
			pluginInstallationCacheKey := helper.PluginInstallationCacheKey(original_plugin_unique_identifier.PluginID(), tenant_id)
			_, _ = cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey)

			if upgradeResponse.IsOriginalPluginDeleted {
				// delete the plugin if no installation left
				manager := plugin_manager.Manager()
				if string(upgradeResponse.DeletedPlugin.InstallType) == string(
					plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL,
				) {
					err = manager.UninstallFromLocal(
						plugin_entities.PluginUniqueIdentifier(upgradeResponse.DeletedPlugin.PluginUniqueIdentifier),
					)
					if err != nil {
						return err
					}
				}
			}

			return nil
		},
	)

	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(response)
}

func FetchPluginInstallationTasks(
	tenant_id string,
	page int,
	page_size int,
) *entities.Response {
	tasks, err := db.GetAll[models.InstallTask](
		db.Equal("tenant_id", tenant_id),
		db.OrderBy("created_at", true),
		db.Page(page, page_size),
	)
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(tasks)
}

func FetchPluginInstallationTask(
	tenant_id string,
	task_id string,
) *entities.Response {
	task, err := db.GetOne[models.InstallTask](
		db.Equal("id", task_id),
		db.Equal("tenant_id", tenant_id),
	)
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(task)
}

func DeletePluginInstallationTask(
	tenant_id string,
	task_id string,
) *entities.Response {
	err := db.DeleteByCondition(
		models.InstallTask{
			Model: models.Model{
				ID: task_id,
			},
			TenantID: tenant_id,
		},
	)

	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(true)
}

func DeleteAllPluginInstallationTasks(
	tenant_id string,
) *entities.Response {
	err := db.DeleteByCondition(
		models.InstallTask{
			TenantID: tenant_id,
		},
	)
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(true)
}

func DeletePluginInstallationItemFromTask(
	tenant_id string,
	task_id string,
	identifier plugin_entities.PluginUniqueIdentifier,
) *entities.Response {
	err := db.WithTransaction(func(tx *gorm.DB) error {
		item, err := db.GetOne[models.InstallTask](
			db.WithTransactionContext(tx),
			db.Equal("id", task_id),
			db.Equal("tenant_id", tenant_id),
			db.WLock(),
		)

		if err != nil {
			return err
		}

		plugins := []models.InstallTaskPluginStatus{}
		for _, plugin := range item.Plugins {
			if plugin.PluginUniqueIdentifier != identifier {
				plugins = append(plugins, plugin)
			}
		}

		successes := 0
		for _, plugin := range plugins {
			if plugin.Status == models.InstallTaskStatusSuccess {
				successes++
			}
		}

		if len(plugins) == successes {
			// delete the task if all plugins are installed successfully
			err = db.Delete(&item, tx)
		} else {
			item.Plugins = plugins
			err = db.Update(&item, tx)
		}

		return err
	})

	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(true)
}

func FetchPluginFromIdentifier(
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
) *entities.Response {
	_, err := db.GetOne[models.Plugin](
		db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
	)
	if err == db.ErrDatabaseNotFound {
		return entities.NewSuccessResponse(false)
	}
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	return entities.NewSuccessResponse(true)
}

func UninstallPlugin(
	tenant_id string,
	plugin_installation_id string,
) *entities.Response {
	// Check if the plugin exists for the tenant
	installation, err := db.GetOne[models.PluginInstallation](
		db.Equal("tenant_id", tenant_id),
		db.Equal("id", plugin_installation_id),
	)
	if err == db.ErrDatabaseNotFound {
		return exception.ErrPluginNotFound().ToResponse()
	}
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	pluginUniqueIdentifier, err := plugin_entities.NewPluginUniqueIdentifier(installation.PluginUniqueIdentifier)
	if err != nil {
		return exception.UniqueIdentifierError(err).ToResponse()
	}

	// get declaration
	declaration, err := helper.CombinedGetPluginDeclaration(
		pluginUniqueIdentifier,
		plugin_entities.PluginRuntimeType(installation.RuntimeType),
	)
	if err != nil {
		return exception.InternalServerError(err).ToResponse()
	}

	// Uninstall the plugin
	deleteResponse, err := curd.UninstallPlugin(
		tenant_id,
		pluginUniqueIdentifier,
		installation.ID,
		declaration,
	)
	if err != nil {
		return exception.InternalServerError(fmt.Errorf("failed to uninstall plugin: %s", err.Error())).ToResponse()
	}

	// invalidate plugin installation cache
	pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginUniqueIdentifier.PluginID(), tenant_id)
	_, _ = cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey)

	if deleteResponse.IsPluginDeleted {
		// delete the plugin if no installation left
		manager := plugin_manager.Manager()
		if deleteResponse.Installation.RuntimeType == string(
			plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL,
		) {
			err = manager.UninstallFromLocal(pluginUniqueIdentifier)
			if err != nil {
				return exception.InternalServerError(fmt.Errorf("failed to uninstall plugin: %s", err.Error())).ToResponse()
			}
		}
	}

	return entities.NewSuccessResponse(true)
}

func FetchPluginReadme(
	tenantId string,
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
	language string,
) *entities.Response {
	if pluginUniqueIdentifier.String() == "" {
		return exception.BadRequestError(errors.New("plugin_unique_identifier is required")).ToResponse()
	}
	if tenantId == "" {
		return exception.BadRequestError(errors.New("tenant_id is required")).ToResponse()
	}

	readmeMap, err := GetPluginReadmeMap(tenantId, pluginUniqueIdentifier)
	if err != nil {
		return exception.InternalServerError(fmt.Errorf("failed to get readme from database: %w", err)).ToResponse()
	}

	if readmeMap == nil || len(readmeMap) == 0 {
		return exception.NotFoundError(errors.New("no readme content available for this plugin")).ToResponse()
	}

	var selectedContent string
	var selectedLanguage string

	if content, exists := readmeMap[language]; exists {
		selectedContent = content
		selectedLanguage = language
	} else if content, exists := readmeMap["en_US"]; exists {
		selectedContent = content
		selectedLanguage = "en_US"
	} else {
		for lang, content := range readmeMap {
			selectedContent = content
			selectedLanguage = lang
			break
		}
	}

	return entities.NewSuccessResponse(map[string]interface{}{
		"content":  selectedContent,
		"language": selectedLanguage,
	})
}