tiny-rdm/backend/services/connection_service.go

package services

import (
	"context"
	"errors"
	"fmt"
	"github.com/redis/go-redis/v9"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"
	. "tinyrdm/backend/storage"
	"tinyrdm/backend/types"
	maputil "tinyrdm/backend/utils/map"
	redis2 "tinyrdm/backend/utils/redis"
)

type connectionService struct {
	ctx     context.Context
	conns   *ConnectionsStorage
	connMap map[string]connectionItem
}

type connectionItem struct {
	rdb        *redis.Client
	ctx        context.Context
	cancelFunc context.CancelFunc
}

type keyItem struct {
	Type string `json:"t"`
}

var connection *connectionService
var onceConnection sync.Once

func Connection() *connectionService {
	if connection == nil {
		onceConnection.Do(func() {
			connection = &connectionService{
				conns:   NewConnections(),
				connMap: map[string]connectionItem{},
			}
		})
	}
	return connection
}

func (c *connectionService) Start(ctx context.Context) {
	c.ctx = ctx
}

func (c *connectionService) Stop(ctx context.Context) {
	for _, item := range c.connMap {
		if item.rdb != nil {
			item.cancelFunc()
			item.rdb.Close()
		}
	}
	c.connMap = map[string]connectionItem{}
}

func (c *connectionService) TestConnection(host string, port int, username, password string) (resp types.JSResp) {
	rdb := redis.NewClient(&redis.Options{
		Addr:     fmt.Sprintf("%s:%d", host, port),
		Username: username,
		Password: password,
	})
	defer rdb.Close()
	if _, err := rdb.Ping(c.ctx).Result(); err != nil && err != redis.Nil {
		resp.Msg = err.Error()
	} else {
		resp.Success = true
	}
	return
}

// ListConnection list all saved connection in local profile
func (c *connectionService) ListConnection() (resp types.JSResp) {
	resp.Success = true
	resp.Data = c.conns.GetConnections()
	return
}

// SaveConnection save connection config to local profile
func (c *connectionService) SaveConnection(param types.Connection, replace bool) (resp types.JSResp) {
	if err := c.conns.UpsertConnection(param, replace); err != nil {
		resp.Msg = err.Error()
	} else {
		resp.Success = true
	}
	return
}

// OpenConnection open redis server connection
func (c *connectionService) OpenConnection(name string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(name, 0)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	// get total database
	config, err := rdb.ConfigGet(ctx, "databases").Result()
	if err != nil {
		resp.Msg = err.Error()
		return
	}
	totaldb, err := strconv.Atoi(config["database"])
	if err != nil {
		totaldb = 16
	}

	// get database info
	res, err := rdb.Info(ctx, "keyspace").Result()
	if err != nil {
		resp.Msg = "list database fail:" + err.Error()
		return
	}
	// Parse all db, response content like below
	var dbs []types.ConnectionDB
	info := c.parseInfo(res)
	for i := 0; i < totaldb; i++ {
		dbName := "db" + strconv.Itoa(i)
		dbInfoStr := info[dbName]
		if len(dbInfoStr) > 0 {
			dbInfo := c.parseDBItemInfo(dbInfoStr)
			dbs = append(dbs, types.ConnectionDB{
				Name:    dbName,
				Keys:    dbInfo["keys"],
				Expires: dbInfo["expires"],
				AvgTTL:  dbInfo["avg_ttl"],
			})
		} else {
			dbs = append(dbs, types.ConnectionDB{
				Name: dbName,
			})
		}
	}

	resp.Success = true
	resp.Data = map[string]any{
		"db": dbs,
	}
	return
}

// CloseConnection close redis server connection
func (c *connectionService) CloseConnection(name string) (resp types.JSResp) {
	item, ok := c.connMap[name]
	if ok {
		delete(c.connMap, name)
		if item.rdb != nil {
			item.cancelFunc()
			item.rdb.Close()
		}
	}
	resp.Success = true
	return
}

// get redis client from local cache or create a new open
// if db >= 0, also switch to db index
func (c *connectionService) getRedisClient(connName string, db int) (*redis.Client, context.Context, error) {
	item, ok := c.connMap[connName]
	var rdb *redis.Client
	var ctx context.Context
	if ok {
		rdb, ctx = item.rdb, item.ctx
	} else {
		connGroups := c.conns.GetConnections()
		var selConn *types.Connection
		for _, connGroup := range connGroups {
			for _, conn := range connGroup.Connections {
				if conn.Name == connName {
					selConn = &conn
					break
				}
			}
		}
		if selConn == nil {
			return nil, nil, errors.New("no match connection connName")
		}

		rdb = redis.NewClient(&redis.Options{
			Addr:        fmt.Sprintf("%s:%d", selConn.Addr, selConn.Port),
			Username:    selConn.Username,
			Password:    selConn.Password,
			ReadTimeout: -1,
		})
		rdb.AddHook(redis2.NewHook(connName))

		if _, err := rdb.Ping(c.ctx).Result(); err != nil && err != redis.Nil {
			return nil, nil, errors.New("can not connect to redis server:" + err.Error())
		}
		var cancelFunc context.CancelFunc
		ctx, cancelFunc = context.WithCancel(c.ctx)
		c.connMap[connName] = connectionItem{
			rdb:        rdb,
			ctx:        ctx,
			cancelFunc: cancelFunc,
		}
	}

	if db >= 0 {
		if err := rdb.Do(ctx, "SELECT", strconv.Itoa(db)).Err(); err != nil {
			return nil, nil, err
		}
	}
	return rdb, ctx, nil
}

// parse command response content which use "redis info"
// # Keyspace\r\ndb0:keys=2,expires=1,avg_ttl=1877111749\r\ndb1:keys=33,expires=0,avg_ttl=0\r\ndb3:keys=17,expires=0,avg_ttl=0\r\ndb5:keys=3,expires=0,avg_ttl=0\r\n
func (c *connectionService) parseInfo(info string) map[string]string {
	parsedInfo := map[string]string{}
	lines := strings.Split(info, "\r\n")
	if len(lines) > 0 {
		for _, line := range lines {
			if !strings.HasPrefix(line, "#") {
				items := strings.SplitN(line, ":", 2)
				if len(items) < 2 {
					continue
				}
				parsedInfo[items[0]] = items[1]
			}
		}
	}
	return parsedInfo
}

// parse db item value, content format like below
// keys=2,expires=1,avg_ttl=1877111749
func (c *connectionService) parseDBItemInfo(info string) map[string]int {
	ret := map[string]int{}
	items := strings.Split(info, ",")
	for _, item := range items {
		kv := strings.SplitN(item, "=", 2)
		if len(kv) > 1 {
			ret[kv[0]], _ = strconv.Atoi(kv[1])
		}
	}
	return ret
}

// OpenDatabase open select database, and list all keys
// @param path contain connection name and db name
func (c *connectionService) OpenDatabase(connName string, db int) (resp types.JSResp) {
	log.Println("open db:" + strconv.Itoa(db))
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	//var keys []string
	keys := map[string]keyItem{}
	var cursor uint64
	for {
		var loadedKey []string
		loadedKey, cursor, err = rdb.Scan(ctx, cursor, "*", 10000).Result()
		if err != nil {
			resp.Msg = err.Error()
			return
		}
		//c.updateDBKey(connName, db, loadedKey)
		for _, k := range loadedKey {
			//t, _ := rdb.Type(ctx, k).Result()
			keys[k] = keyItem{Type: "t"}
		}
		//keys = append(keys, loadedKey...)
		// no more loadedKey
		if cursor == 0 {
			break
		}
	}

	resp.Success = true
	resp.Data = map[string]any{
		"keys": keys,
	}
	return
}

// GetKeyValue get value by key
func (c *connectionService) GetKeyValue(connName string, db int, key string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var keyType string
	var dur time.Duration
	keyType, err = rdb.Type(ctx, key).Result()
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var ttl int64
	if dur, err = rdb.TTL(ctx, key).Result(); err != nil {
		ttl = -1
	} else {
		if dur < 0 {
			ttl = -1
		} else {
			ttl = int64(dur.Seconds())
		}
	}

	var value any
	var cursor uint64
	switch strings.ToLower(keyType) {
	case "string":
		value, err = rdb.Get(ctx, key).Result()
	case "list":
		value, err = rdb.LRange(ctx, key, 0, -1).Result()
	case "hash":
		//value, err = rdb.HGetAll(ctx, key).Result()
		items := map[string]string{}
		for {
			var loadedVal []string
			loadedVal, cursor, err = rdb.HScan(ctx, key, cursor, "*", 10000).Result()
			if err != nil {
				resp.Msg = err.Error()
				return
			}
			for i := 0; i < len(loadedVal); i += 2 {
				items[loadedVal[i]] = loadedVal[i+1]
			}
			if cursor == 0 {
				break
			}
		}
		value = items
	case "set":
		//value, err = rdb.SMembers(ctx, key).Result()
		items := []string{}
		for {
			var loadedKey []string
			loadedKey, cursor, err = rdb.SScan(ctx, key, cursor, "*", 10000).Result()
			if err != nil {
				resp.Msg = err.Error()
				return
			}
			items = append(items, loadedKey...)
			if cursor == 0 {
				break
			}
		}
		value = items
	case "zset":
		//value, err = rdb.ZRangeWithScores(ctx, key, 0, -1).Result()
		var items []types.ZSetItem
		for {
			var loadedVal []string
			loadedVal, cursor, err = rdb.ZScan(ctx, key, cursor, "*", 10000).Result()
			if err != nil {
				resp.Msg = err.Error()
				return
			}
			var score float64
			for i := 0; i < len(loadedVal); i += 2 {
				if score, err = strconv.ParseFloat(loadedVal[i+1], 64); err == nil {
					items = append(items, types.ZSetItem{
						Value: loadedVal[i],
						Score: score,
					})
				}
			}
			if cursor == 0 {
				break
			}
		}
		value = items
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}
	resp.Success = true
	resp.Data = map[string]any{
		"type":  keyType,
		"ttl":   ttl,
		"value": value,
	}
	return
}

// SetKeyValue set value by key
func (c *connectionService) SetKeyValue(connName string, db int, key, keyType string, value any, ttl int64) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var expiration time.Duration
	if ttl < 0 {
		expiration = redis.KeepTTL
	} else {
		expiration = time.Duration(ttl) * time.Second
	}
	switch strings.ToLower(keyType) {
	case "string":
		if str, ok := value.(string); !ok {
			resp.Msg = "invalid string value"
			return
		} else {
			_, err = rdb.Set(ctx, key, str, expiration).Result()
		}
	case "list":
		if strs, ok := value.([]any); !ok {
			resp.Msg = "invalid list value"
			return
		} else {
			_, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				pipe.LPush(ctx, key, strs...)
				if expiration > 0 {
					pipe.Expire(ctx, key, expiration)
				}
				return nil
			})
		}
	case "hash":
		if strs, ok := value.([]any); !ok {
			resp.Msg = "invalid hash value"
			return
		} else {
			_, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				if len(strs) > 1 {
					for i := 0; i < len(strs); i += 2 {
						pipe.HSetNX(ctx, key, strs[i].(string), strs[i+1])
					}
				} else {
					pipe.HSet(ctx, key)
				}
				if expiration > 0 {
					pipe.Expire(ctx, key, expiration)
				}
				return nil
			})
		}
	case "set":
		if strs, ok := value.([]any); !ok || len(strs) <= 0 {
			resp.Msg = "invalid set value"
			return
		} else {
			_, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				if len(strs) > 1 {
					for _, str := range strs {
						pipe.SAdd(ctx, key, str.(string))
					}
				} else {
					pipe.SAdd(ctx, key)
				}
				if expiration > 0 {
					pipe.Expire(ctx, key, expiration)
				}
				return nil
			})
		}
	case "zset":
		if strs, ok := value.([]any); !ok || len(strs) <= 0 {
			resp.Msg = "invalid zset value"
			return
		} else {
			_, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
				var members []redis.Z
				for i := 0; i < len(strs); i += 2 {
					members = append(members, redis.Z{
						Score:  strs[i].(float64),
						Member: strs[i+1],
					})
				}

				if len(members) > 0 {
					pipe.ZAdd(ctx, key, members...)
				} else {
					pipe.ZAdd(ctx, key)
				}
				if expiration > 0 {
					pipe.Expire(ctx, key, expiration)
				}
				return nil
			})
		}
	}

	if err != nil {
		resp.Msg = err.Error()
		return
	}
	resp.Success = true
	resp.Data = map[string]any{
		"value": value,
	}
	return
}

// SetHashValue set hash field
func (c *connectionService) SetHashValue(connName string, db int, key, field, newField, value string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var removedField []string
	updatedField := map[string]string{}
	if len(field) <= 0 {
		// old filed is empty, add new field
		_, err = rdb.HSet(ctx, key, newField, value).Result()
		updatedField[newField] = value
	} else if len(newField) <= 0 {
		// new field is empty, delete old field
		_, err = rdb.HDel(ctx, key, field, value).Result()
		removedField = append(removedField, field)
	} else if field == newField {
		// replace field
		_, err = rdb.HSet(ctx, key, newField, value).Result()
		updatedField[newField] = value
	} else {
		// remove old field and add new field
		if _, err = rdb.HDel(ctx, key, field).Result(); err != nil {
			resp.Msg = err.Error()
			return
		}
		_, err = rdb.HSet(ctx, key, newField, value).Result()
		removedField = append(removedField, field)
		updatedField[newField] = value
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	resp.Data = map[string]any{
		"removed": removedField,
		"updated": updatedField,
	}
	return
}

// AddHashField add or update hash field
func (c *connectionService) AddHashField(connName string, db int, key string, action int, fieldItems []any) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	updated := map[string]any{}
	switch action {
	case 1:
		// ignore duplicated fields
		for i := 0; i < len(fieldItems); i += 2 {
			_, err = rdb.HSetNX(ctx, key, fieldItems[i].(string), fieldItems[i+1]).Result()
			if err == nil {
				updated[fieldItems[i].(string)] = fieldItems[i+1]
			}
		}
	default:
		// overwrite duplicated fields
		_, err = rdb.HSet(ctx, key, fieldItems...).Result()
		for i := 0; i < len(fieldItems); i += 2 {
			updated[fieldItems[i].(string)] = fieldItems[i+1]
		}
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	resp.Data = map[string]any{
		"updated": updated,
	}
	return
}

// AddListItem add item to list or remove from it
func (c *connectionService) AddListItem(connName string, db int, key string, action int, items []any) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var leftPush, rightPush []any
	switch action {
	case 0:
		// push to head
		_, err = rdb.LPush(ctx, key, items...).Result()
		leftPush = append(leftPush, items...)
	default:
		// append to tail
		_, err = rdb.RPush(ctx, key, items...).Result()
		rightPush = append(rightPush, items...)
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	resp.Data = map[string]any{
		"left":  leftPush,
		"right": rightPush,
	}
	return
}

// SetListItem update or remove list item by index
func (c *connectionService) SetListItem(connName string, db int, key string, index int64, value string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var removed []int64
	updated := map[int64]string{}
	if len(value) <= 0 {
		// remove from list
		err = rdb.LSet(ctx, key, index, "---VALUE_REMOVED_BY_TINY_RDM---").Err()
		if err != nil {
			resp.Msg = err.Error()
			return
		}

		err = rdb.LRem(ctx, key, 1, "---VALUE_REMOVED_BY_TINY_RDM---").Err()
		if err != nil {
			resp.Msg = err.Error()
			return
		}
		removed = append(removed, index)
	} else {
		// replace index value
		err = rdb.LSet(ctx, key, index, value).Err()
		if err != nil {
			resp.Msg = err.Error()
			return
		}
		updated[index] = value
	}

	resp.Success = true
	resp.Data = map[string]any{
		"removed": removed,
		"updated": updated,
	}
	return
}

// SetSetItem add members to set or remove from set
func (c *connectionService) SetSetItem(connName string, db int, key string, remove bool, members []any) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	if remove {
		_, err = rdb.SRem(ctx, key, members...).Result()
	} else {
		_, err = rdb.SAdd(ctx, key, members...).Result()
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	return
}

// UpdateSetItem replace member of set
func (c *connectionService) UpdateSetItem(connName string, db int, key, value, newValue string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	_, _ = rdb.SRem(ctx, key, value).Result()
	_, err = rdb.SAdd(ctx, key, newValue).Result()
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	return
}

// UpdateZSetValue update value of sorted set member
func (c *connectionService) UpdateZSetValue(connName string, db int, key, value, newValue string, score float64) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	updated := map[string]any{}
	var removed []string
	if len(newValue) <= 0 {
		// blank new value, delete value
		_, err = rdb.ZRem(ctx, key, value).Result()
		if err == nil {
			removed = append(removed, value)
		}
	} else if newValue == value {
		// update score only
		_, err = rdb.ZAdd(ctx, key, redis.Z{
			Score:  score,
			Member: value,
		}).Result()
	} else {
		// remove old value and add new one
		_, err = rdb.ZRem(ctx, key, value).Result()
		if err == nil {
			removed = append(removed, value)
		}

		_, err = rdb.ZAdd(ctx, key, redis.Z{
			Score:  score,
			Member: newValue,
		}).Result()
		if err == nil {
			updated[newValue] = score
		}
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	resp.Data = map[string]any{
		"updated": updated,
		"removed": removed,
	}
	return
}

// AddZSetValue add item to sorted set
func (c *connectionService) AddZSetValue(connName string, db int, key string, action int, valueScore map[string]float64) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	members := maputil.ToSlice(valueScore, func(k string) redis.Z {
		return redis.Z{
			Score:  valueScore[k],
			Member: k,
		}
	})

	switch action {
	case 1:
		// ignore duplicated fields
		_, err = rdb.ZAddNX(ctx, key, members...).Result()
	default:
		// overwrite duplicated fields
		_, err = rdb.ZAdd(ctx, key, members...).Result()
	}
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	return
}

// SetKeyTTL set ttl of key
func (c *connectionService) SetKeyTTL(connName string, db int, key string, ttl int64) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	var expiration time.Duration
	if ttl < 0 {
		if err = rdb.Persist(ctx, key).Err(); err != nil {
			resp.Msg = err.Error()
			return
		}
	} else {
		expiration = time.Duration(ttl) * time.Second
		if err = rdb.Expire(ctx, key, expiration).Err(); err != nil {
			resp.Msg = err.Error()
			return
		}
	}

	resp.Success = true
	return
}

// RemoveKey remove redis key
func (c *connectionService) RemoveKey(connName string, db int, key string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	rmCount, err := rdb.Del(ctx, key).Result()
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	resp.Data = map[string]any{
		"effect_count": rmCount,
	}
	return
}

// RenameKey rename key
func (c *connectionService) RenameKey(connName string, db int, key, newKey string) (resp types.JSResp) {
	rdb, ctx, err := c.getRedisClient(connName, db)
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	_, err = rdb.RenameNX(ctx, key, newKey).Result()
	if err != nil {
		resp.Msg = err.Error()
		return
	}

	resp.Success = true
	return
}

// update or insert key info to database
//func (c *connectionService) updateDBKey(connName string, db int, keys []string, separator string) {
//	dbStruct := map[string]any{}
//	for _, key := range keys {
//		keyPart := strings.Split(key, separator)
//		prefixLen := len(keyPart)-1
//		if prefixLen > 0 {
//			for i := 0; i < prefixLen; i++ {
//				if dbStruct[keyPart[i]]
//				keyPart[i]
//			}
//		}
//		log.Println("key", key)
//	}
//}