feat(service): add reset mode

README.md

@@ -122,6 +122,7 @@ Variable Name  | Description                                              | Defa
 
 Variable Name  | Description                                              | Default
 -------------- | -------------------------------------------------------- | -------------------------------------------------------------
+`RESET_MODE`   | Enable Auto Reset Status, **NOTE: In cluster**           | `true`
 `SCHEDULE`     | Only Single Node. **Not support cluster**                | `true`
 `EMQX_AUTH`    | Use Emqx redis auth plugin. If Mosquitto, Set `false`    | `false`
 `LUA_FILE`     | Task lua > `LUA_FILE` > Default(`luavm/lua/default.lua`) | `default.lua`

app/config/config.go

@@ -29,6 +29,7 @@ type Config struct {
 	ListenAddr string `env:"LISTEN_ADDR"`
 
 	// Feature Flags
+	ResetMode   bool   `env:"RESET_MODE"`
 	Schedule    bool   `env:"SCHEDULE"`
 	EmqxAuth    bool   `env:"EMQX_AUTH"`
 	LuaFilePath string `env:"LUA_FILE"`
@@ -63,6 +64,7 @@ func DefaultConfig() *Config {
 		ApiMqttWs:  "ws://localhost:8083/mqtt",
 		ListenAddr: "0.0.0.0:8000",
 
+		ResetMode:   true,
 		Schedule:    true,
 		EmqxAuth:    false,
 		LuaFilePath: "default.lua",

app/daemon/daemon.go

@@ -65,6 +65,10 @@ func NewHandler(ctx context.Context, cfg *config.Config) http.Handler {
 	go rpcServer.Run(ctx)
 
 	srv := service.NewService(cfg, orm, rdb, ofs)
+	if cfg.ResetMode {
+		srv.Reset(ctx)
+	}
+
 	if cfg.Schedule {
 		go srv.RunSchedule(ctx)
 	}

app/service/reset.go

@@ -0,0 +1,27 @@
+package service
+
+import (
+	"context"
+)
+
+func (s *Service) Reset(ctx context.Context) error {
+	prefix := "luavm.*"
+	var cursor uint64
+	for {
+		var keys []string
+		var err error
+		keys, cursor, err = s.rdb.Scan(ctx, cursor, prefix, 0).Result()
+		if err != nil {
+			panic(err)
+		}
+
+		for _, key := range keys {
+			s.rdb.Del(ctx, key)
+		}
+
+		if cursor == 0 {
+			break
+		}
+	}
+	return nil
+}

app/service/reset_test.go

@@ -0,0 +1,67 @@
+package service
+
+import (
+	"context"
+	"testing"
+
+	"sb.im/gosd/app/config"
+	"sb.im/gosd/app/storage"
+
+	"github.com/redis/go-redis/v9"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestReset(t *testing.T) {
+	id := "test_id"
+	srv := testNewService(t)
+
+	if err := srv.LockTaskSet(id); err != nil {
+		t.Error(err)
+	}
+	if _, err := srv.LockTaskGet(id); err != nil {
+		t.Error(err)
+	}
+
+	if err := srv.LockNodeSet(id); err != nil {
+		t.Error(err)
+	}
+	if _, err := srv.LockNodeGet(id); err != nil {
+		t.Error(err)
+	}
+
+	srv.LockTaskGet(id)
+	if err := srv.Reset(context.Background()); err != nil {
+		t.Error(err)
+	}
+
+	if key, err := srv.LockTaskGet(id); err == nil {
+		t.Error(key)
+	}
+	if key, err := srv.LockNodeGet(id); err == nil {
+		t.Error(key)
+	}
+}
+
+func testNewService(t *testing.T) *Service {
+	cfg := config.Parse()
+	orm, err := gorm.Open(postgres.Open(cfg.DatabaseURL), nil)
+	if err != nil {
+		t.Error(err)
+	}
+
+	redisOpt, err := redis.ParseURL(cfg.RedisURL)
+	if err != nil {
+		t.Error(err)
+	}
+	rdb := redis.NewClient(redisOpt)
+
+	// Enable Redis Events
+	// K: store
+	// Ex: luavm
+	rdb.ConfigSet(context.Background(), "notify-keyspace-events", "$KEx")
+
+	ofs := storage.NewStorage(cfg.StorageURL)
+
+	return NewService(cfg, orm, rdb, ofs)
+}