fix: add unique index for plugin installation to prevent duplicate entries (#528)

Author: Mairuis

internal/service/install_plugin.go

@@ -3,7 +3,6 @@ package service
 import (
 	"errors"
 	"fmt"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	controlpanel "github.com/langgenius/dify-plugin-daemon/internal/core/control_panel"
@@ -105,49 +104,21 @@ func InstallMultiplePluginsToTenant(
 		})
 	}
 
-	// acquire locks to prevent concurrent installation of the same plugin
-	lockKeys := make(map[string]string, len(pluginUniqueIdentifiers))
-	releaseLocks := func() {
-		for _, key := range lockKeys {
-			if err := cache.Unlock(key); err != nil {
-				log.Error("failed to unlock key %s: %v", key, err)
-			}
-		}
-	}
-	for _, pluginUniqueIdentifier := range pluginUniqueIdentifiers {
-		lockKey := fmt.Sprintf("plugin:install:%s:%s", tenantId, pluginUniqueIdentifier.PluginID())
-		if err := cache.Lock(lockKey, time.Minute*10, time.Millisecond*100); err != nil {
-			releaseLocks()
-			if err == cache.ErrLockTimeout {
-				return exception.BadRequestError(errors.New("plugin installation is already in progress")).ToResponse()
-			}
-			return exception.InternalServerError(err).ToResponse()
-		}
-		lockKeys[pluginUniqueIdentifier.PluginID()] = lockKey
-	}
-
 	// create tasks for each plugin
 	statuses := buildTaskStatuses(pluginUniqueIdentifiers, declarations)
 	taskRegistry, err := createInstallTasks(tenants, statuses)
 	if err != nil {
-		releaseLocks()
 		return exception.InternalServerError(err).ToResponse()
 	}
 	taskIDs := taskRegistry.IDs()
 
 	for _, job := range jobs {
 		jobCopy := job
-		lockKey := lockKeys[jobCopy.Identifier.PluginID()]
 		// start a new goroutine to install the plugin
 		routine.Submit(routinepkg.Labels{
 			routinepkg.RoutineLabelKeyModule: "service",
 			routinepkg.RoutineLabelKeyMethod: "InstallPlugin",
 		}, func() {
-			defer func() {
-				if err := cache.Unlock(lockKey); err != nil {
-					log.Error("failed to unlock key %s: %v", lockKey, err)
-				}
-			}()
 			tasks.ProcessInstallJob(
 				manager,
 				tenants,
@@ -288,15 +259,6 @@ func UpgradePlugin(
 		return exception.InternalServerError(err).ToResponse()
 	}
 
-	// acquire lock to prevent concurrent upgrade of the same plugin
-	lockKey := fmt.Sprintf("plugin:install:%s:%s", tenantId, newPluginUniqueIdentifier.PluginID())
-	if err := cache.Lock(lockKey, time.Minute*10, time.Millisecond*100); err != nil {
-		if err == cache.ErrLockTimeout {
-			return exception.BadRequestError(errors.New("plugin installation is already in progress")).ToResponse()
-		}
-		return exception.InternalServerError(err).ToResponse()
-	}
-
 	// construct tenant jobs
 	tenants := []string{tenantId}
 
@@ -315,9 +277,6 @@ func UpgradePlugin(
 
 	taskRegistry, err := createInstallTasks(tenants, statuses)
 	if err != nil {
-		if unlockErr := cache.Unlock(lockKey); unlockErr != nil {
-			log.Error("failed to unlock key %s: %v", lockKey, unlockErr)
-		}
 		return exception.InternalServerError(err).ToResponse()
 	}
 
@@ -327,11 +286,6 @@ func UpgradePlugin(
 		routinepkg.RoutineLabelKeyModule: "service",
 		routinepkg.RoutineLabelKeyMethod: "UpgradePlugin",
 	}, func() {
-		defer func() {
-			if err := cache.Unlock(lockKey); err != nil {
-				log.Error("failed to unlock key %s: %v", lockKey, err)
-			}
-		}()
 		tasks.ProcessUpgradeJob(
 			manager,
 			tenants,

internal/types/models/installation.go

@@ -4,8 +4,8 @@ type PluginInstallationStatus string
 
 type PluginInstallation struct {
 	Model
-	TenantID               string         `json:"tenant_id" gorm:"index;type:uuid;"`
-	PluginID               string         `json:"plugin_id" gorm:"index;size:255"`
+	TenantID               string         `json:"tenant_id" gorm:"index;uniqueIndex:idx_tenant_plugin;type:uuid;"`
+	PluginID               string         `json:"plugin_id" gorm:"index;uniqueIndex:idx_tenant_plugin;size:255"`
 	PluginUniqueIdentifier string         `json:"plugin_unique_identifier" gorm:"index;size:255"`
 	RuntimeType            string         `json:"runtime_type" gorm:"size:127"`
 	EndpointsSetups        int            `json:"endpoints_setups"`