feat: remote plugin disconnect clean the plugin cache (#667)

Author: fatelei

internal/service/install_service/controlpanel.go

@@ -9,6 +9,8 @@ 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/plugin_entities"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache/helper"
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/log"
 )
 
@@ -72,6 +74,11 @@ func (l *InstallListener) OnDebuggingRuntimeConnected(runtime *debugging_runtime
 			return
 		}
 		installation = existingInstallation
+		
+		key := helper.PluginInstallationCacheKey(pluginID, runtime.TenantId())
+		if _, delErr := cache.AutoDelete[models.PluginInstallation](key); delErr != nil {
+			log.Warn("failed to invalidate plugin installation cache", "key", key, "error", delErr)
+		}
 	}
 
 	// FIXME(Yeuoly): temporary solution for managing plugin installation model in DB

internal/types/models/curd/atomic.go

@@ -226,6 +226,13 @@ func InstallPlugin(
 		return nil, nil, err
 	}
 
+	// Invalidate plugin installation cache to avoid stale reads
+	pluginID := pluginToBeReturns.PluginID
+	pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginID, tenantId)
+	if _, delErr := cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey); delErr != nil {
+		log.Warn("failed to clear plugin installation cache", "key", pluginInstallationCacheKey, "error", delErr)
+	}
+
 	return pluginToBeReturns, installationToBeReturns, nil
 }
 
@@ -660,8 +667,8 @@ func UpgradePlugin(
 	if err != nil {
 		return nil, err
 	}
-	pluginId := newPluginUniqueIdentifier.PluginID()                                    // get the pluginId
-	pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginId, tenantId) // make cache key
+	pluginID := newPluginUniqueIdentifier.PluginID()                                    // get the pluginId
+	pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginID, tenantId) // make cache key
 	if _, err = cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey); err != nil {
 		return nil, err
 	}

internal/types/models/curd/cache_invalidation_test.go

@@ -0,0 +1,83 @@
+package curd
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"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/entities/plugin_entities"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache/helper"
+	"github.com/stretchr/testify/require"
+)
+
+// Test that InstallPlugin invalidates PluginInstallation cache key so that
+// subsequent reads fall back to DB instead of returning stale data.
+func TestInstallPlugin_InvalidateInstallationCache(t *testing.T) {
+	// Init Redis
+	require.NoError(t, cache.InitRedisClient("127.0.0.1:6379", "", "difyai123456", false, 0, nil))
+	t.Cleanup(func() { _ = cache.Close() })
+
+	// Init DB
+	cfg := &app.Config{
+		DBType:     app.DB_TYPE_POSTGRESQL,
+		DBUsername: "postgres",
+		DBPassword: "difyai123456",
+		DBHost:     "localhost",
+		DBPort:     5432,
+		DBDatabase: "dify_plugin_daemon",
+		DBSslMode:  "disable",
+	}
+	cfg.SetDefault()
+	db.Init(cfg)
+	t.Cleanup(db.Close)
+
+	tenantID := uuid.NewString()
+	pluginName := "cache_invalidate_" + uuid.NewString()
+	checksum := strings.ReplaceAll(uuid.NewString(), "-", "")
+	if len(checksum) > 32 {
+		checksum = checksum[:32]
+	}
+
+	identifier, err := plugin_entities.NewPluginUniqueIdentifier("tester/" + pluginName + ":1.0.0.0@" + checksum)
+	require.NoError(t, err)
+	pluginID := identifier.PluginID()
+
+	// Seed a stale cache value for the installation key
+	key := helper.PluginInstallationCacheKey(pluginID, tenantID)
+	require.NoError(t, cache.AutoSet[models.PluginInstallation](key, models.PluginInstallation{
+		Model:                  models.Model{ID: "OLD"},
+		PluginID:               pluginID,
+		PluginUniqueIdentifier: identifier.String(),
+		TenantID:               tenantID,
+		RuntimeType:            string(plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL),
+	}))
+	// Perform install which should invalidate the cache key
+	_, _, err = InstallPlugin(
+		tenantID,
+		identifier,
+		plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL,
+		&plugin_entities.PluginDeclaration{},
+		"unittest",
+		map[string]any{"from": "test"},
+	)
+	require.NoError(t, err)
+
+	// Read using AutoGetWithGetter — should miss cache, call getter (DB), then set fresh cache
+	inst, err := cache.AutoGetWithGetter(key, func() (*models.PluginInstallation, error) {
+		v, e := db.GetOne[models.PluginInstallation](
+			db.Equal("plugin_id", pluginID),
+			db.Equal("tenant_id", tenantID),
+		)
+		if e != nil {
+			return nil, e
+		}
+		return &v, nil
+	})
+	require.NoError(t, err)
+	require.NotNil(t, inst)
+	require.NotEqual(t, "OLD", inst.ID, "cache should have been invalidated and refetched from DB")
+}