mirror of
https://github.com/langgenius/dify-plugin-daemon.git
synced 2026-07-22 09:45:27 -04:00
888ad788bc
* refactor: introduce local plugin control panel and cleanup environment setup process * fix: args * refactor: new local runtime * temp: stash work for refactor on RemotePluginServer * refactor: unify local runtime lifetime and sperate init environment process * chore: add missing files * stash * refactor: local plugin lifetime control * refactor: complete installation process of control panel * refactor: adapt service layer to new controlpanel * refactor: pluginManager.Install * fix: add routine wrap to InstallServerless, avoid blocking main thread * feat: reinstall serverless runtime * chore: add comments to Reinstall and update confusing naming * refactor: unify install plugin service * refactor: add labels to debugging runtime * refactor: add getters to plugin manager * refactor: split install service to decode/install_task/install service * ??? * refactor: adapt controllers * refactor: session write * refactor: session runtime * Refine install task orchestration (#501) * refactor: installing task * refactor cluster management, decouple lifetime management and cluster * fix cli test command * fix: cleanup TODO comments and implement GracefulStop for instance * feat: add logger to control panel * fix: multiple nil references * refactor: better lifetime control * refactor: better cycle interval * fix(LocalPluginRuntime): prevent returning err when it's not error * fix: avoid adding empty PipExtraArgs * fix: missing errors in Environment init * fix: add truncateMessage to avoid db explosion * cleanup: better lifecycle management * fix: init status at the beginning of installation * optimize: GracefulStop for pluginInstance * refactor: tests * refactor: centralize routine labels (#504) * cleanup: RoutineKey * fix: init routine pool * fix: correctly handle cluster register error * fix: memory leak * fix: add \n to instance write * fix(installer.go): set success to true after succeed for defer func * refactor * fix: missing cwd in testutils * fix: scaleup default runtime nums to 1 when testing * fix: localruntime appconfig in testing module * Update internal/core/local_runtime/load_balancing.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * fix: more efficiency implement in installer_local.go * fix: returns after failing in onDebuggingRuntimeDisconnected * fix: returns after failing in onDebuggingRuntimeDisconnected * fix: splits tests * refactor: naming * refactor: manifest.VersionX * fix: adapt SetDefault to tests * fix: enforce use constants in DBType * fix: generate * fix: linter * cleanup tests * refactor: change package to * cleanup: useless codes * Update internal/cluster/plugin.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * cleanup * refactor: decouple connection_key management from debugging_time * refactor: confused naming * feat: recycle resources to adapt to https://github.com/langgenius/dify-plugin-daemon/pull/500 * refactor: confusing redirecting * fix: support get serverless runtime * fix: race condition in Launching * fix: avoid ManifestValidate in first step of debugging handshake * fix: adding ReleaseAllLocks to finalizers * wtf: what a beautiful code * refactor: rename Stream.Async to Stream.Process * fix: kill process if daed instance was detected * fix: correctly handle failures * fix: consistence of difference interfaces * fix: add stacktrace to panic * fix: only trigger once event * fix: ensure plugin runtime was shutdown * feat: cleanup install tasks * fix: add scale logs --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
177 lines
4.0 KiB
Go
177 lines
4.0 KiB
Go
package persistence
|
|
|
|
import (
|
|
"encoding/hex"
|
|
"fmt"
|
|
"path"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/langgenius/dify-plugin-daemon/internal/db"
|
|
"github.com/langgenius/dify-plugin-daemon/internal/types/models"
|
|
"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
|
|
)
|
|
|
|
type Persistence struct {
|
|
maxStorageSize int64
|
|
|
|
storage PersistenceStorage
|
|
}
|
|
|
|
const (
|
|
CACHE_KEY_PREFIX = "persistence:cache"
|
|
)
|
|
|
|
func (c *Persistence) getCacheKey(tenantId string, pluginId string, key string) string {
|
|
return fmt.Sprintf("%s:%s:%s:%s", CACHE_KEY_PREFIX, tenantId, pluginId, key)
|
|
}
|
|
|
|
func (c *Persistence) checkPathTraversal(key string) error {
|
|
key = path.Clean(key)
|
|
if strings.Contains(key, "..") || strings.Contains(key, "//") || strings.Contains(key, "\\") {
|
|
return fmt.Errorf("invalid key: path traversal attempt detected")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *Persistence) Save(tenantId string, pluginId string, maxSize int64, key string, data []byte) error {
|
|
if err := c.checkPathTraversal(key); err != nil {
|
|
return err
|
|
}
|
|
|
|
if len(key) > 256 {
|
|
return fmt.Errorf("key length must be less than 256 characters")
|
|
}
|
|
|
|
if maxSize == -1 {
|
|
maxSize = c.maxStorageSize
|
|
}
|
|
|
|
if err := c.storage.Save(tenantId, pluginId, key, data); err != nil {
|
|
return err
|
|
}
|
|
|
|
allocatedSize := int64(len(data))
|
|
|
|
storage, err := db.GetOne[models.TenantStorage](
|
|
db.Equal("tenant_id", tenantId),
|
|
db.Equal("plugin_id", pluginId),
|
|
)
|
|
if err != nil {
|
|
if allocatedSize > c.maxStorageSize || allocatedSize > maxSize {
|
|
return fmt.Errorf("allocated size is greater than max storage size")
|
|
}
|
|
|
|
if err == db.ErrDatabaseNotFound {
|
|
storage = models.TenantStorage{
|
|
TenantID: tenantId,
|
|
PluginID: pluginId,
|
|
Size: allocatedSize,
|
|
}
|
|
if err := db.Create(&storage); err != nil {
|
|
return err
|
|
}
|
|
} else {
|
|
return err
|
|
}
|
|
} else {
|
|
if allocatedSize+storage.Size > maxSize || allocatedSize+storage.Size > c.maxStorageSize {
|
|
return fmt.Errorf("allocated size is greater than max storage size")
|
|
}
|
|
|
|
err = db.Run(
|
|
db.Model(&models.TenantStorage{}),
|
|
db.Equal("tenant_id", tenantId),
|
|
db.Equal("plugin_id", pluginId),
|
|
db.Inc(map[string]int64{"size": allocatedSize}),
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// delete from cache
|
|
if _, err = cache.Del(c.getCacheKey(tenantId, pluginId, key)); err == cache.ErrNotFound {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// TODO: raises specific error to avoid confusion
|
|
func (c *Persistence) Load(tenantId string, pluginId string, key string) ([]byte, error) {
|
|
if err := c.checkPathTraversal(key); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// check if the key exists in cache
|
|
h, err := cache.GetString(c.getCacheKey(tenantId, pluginId, key))
|
|
if err != nil && err != cache.ErrNotFound {
|
|
return nil, err
|
|
}
|
|
if err == nil {
|
|
return hex.DecodeString(h)
|
|
}
|
|
|
|
// load from storage
|
|
data, err := c.storage.Load(tenantId, pluginId, key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// add to cache
|
|
cache.Store(c.getCacheKey(tenantId, pluginId, key), hex.EncodeToString(data), time.Minute*5)
|
|
|
|
return data, nil
|
|
}
|
|
|
|
func (c *Persistence) Delete(tenantId string, pluginId string, key string) (int64, error) {
|
|
// delete from cache and storage
|
|
deletedNum, err := cache.Del(c.getCacheKey(tenantId, pluginId, key))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
// state size
|
|
size, err := c.storage.StateSize(tenantId, pluginId, key)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
err = c.storage.Delete(tenantId, pluginId, key)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
// update storage size
|
|
err = db.Run(
|
|
db.Model(&models.TenantStorage{}),
|
|
db.Equal("tenant_id", tenantId),
|
|
db.Equal("plugin_id", pluginId),
|
|
db.Dec(map[string]int64{"size": size}),
|
|
)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
return deletedNum, nil
|
|
}
|
|
|
|
func (c *Persistence) Exist(tenantId string, pluginId string, key string) (int64, error) {
|
|
existNum, err := cache.Exist(c.getCacheKey(tenantId, pluginId, key))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if existNum > 0 {
|
|
return existNum, nil
|
|
}
|
|
|
|
isExist, err := c.storage.Exists(tenantId, pluginId, key)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if isExist {
|
|
return 1, nil
|
|
}
|
|
return 0, nil
|
|
}
|