Repository: lf-edge/ekuiper
Pull request #2962 (2024-06-28 07:50:37)

feat: support httppull incremental


Patch:
diff --git a/internal/io/sql/client/dburl.go b/extension/sql/client/dburl.go
similarity index 100%
rename from internal/io/sql/client/dburl.go
rename to extension/sql/client/dburl.go
diff --git a/internal/io/sql/sink.go b/extension/sql/sink.go
similarity index 98%
rename from internal/io/sql/sink.go
rename to extension/sql/sink.go
index 3ac421a8..39f3e56a 100644
--- a/internal/io/sql/sink.go
+++ b/extension/sql/sink.go
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build full
-// +build full
-
 package sql
 
 import (
@@ -23,7 +20,7 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
diff --git a/internal/io/sql/source.go b/extension/sql/source.go
similarity index 98%
rename from internal/io/sql/source.go
rename to extension/sql/source.go
index 5ff91f9d..7c45a023 100644
--- a/internal/io/sql/source.go
+++ b/extension/sql/source.go
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build full
-// +build full
-
 package sql
 
 import (
@@ -28,7 +25,7 @@ import (
 	"github.com/pingcap/failpoint"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/sqldatabase/sqlgen"
diff --git a/internal/io/sql/testx/mysql_server.go b/extension/sql/testx/mysql_server.go
similarity index 100%
rename from internal/io/sql/testx/mysql_server.go
rename to extension/sql/testx/mysql_server.go
diff --git a/internal/io/video/source.go b/extension/video/source.go
similarity index 100%
rename from internal/io/video/source.go
rename to extension/video/source.go
diff --git a/internal/io/video/source_test.go b/extension/video/source_test.go
similarity index 100%
rename from internal/io/video/source_test.go
rename to extension/video/source_test.go
diff --git a/go.mod b/go.mod
index 23cf0089..25f9687b 100644
--- a/go.mod
+++ b/go.mod
@@ -26,7 +26,7 @@ require (
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1
 	github.com/klauspost/compress v1.17.7
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible
-	github.com/lf-edge/ekuiper/contract/v2 v2.0.0-alpha.2
+	github.com/lf-edge/ekuiper/contract/v2 v2.0.0-alpha.3
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/mochi-mqtt/server/v2 v2.6.4
 	github.com/modern-go/reflect2 v1.0.2
diff --git a/internal/binder/io/ext_full.go b/internal/binder/io/ext_full.go
index 8d84478a..6f79423d 100644
--- a/internal/binder/io/ext_full.go
+++ b/internal/binder/io/ext_full.go
@@ -18,9 +18,9 @@ package io
 
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
-	"github.com/lf-edge/ekuiper/v2/internal/io/video"
+	"github.com/lf-edge/ekuiper/v2/extension/sql"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extension/video"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
diff --git a/internal/io/connection.go b/internal/io/connection.go
deleted file mode 100644
index f366cd7e..00000000
--- a/internal/io/connection.go
+++ /dev/null
@@ -1,41 +0,0 @@
-// Copyright 2024 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package io
-
-import "github.com/lf-edge/ekuiper/contract/v2/api"
-
-type ConnStatus int
-
-const (
-	Connecting ConnStatus = iota
-	Connected
-	Disconnect
-	Subscribed
-	SubError
-)
-
-type MessageClient interface {
-	Subscribe(c api.StreamContext, subChan []TopicChannel, messageErrors chan error, params map[string]interface{}) error
-	Publish(c api.StreamContext, topic string, message []byte, params map[string]interface{}) error
-	Ping() error
-}
-
-// TopicChannel is the data structure for subscriber
-type TopicChannel struct {
-	// Topic for subscriber to filter on if any
-	Topic string
-	// Messages is the returned message channel for the subscriber
-	Messages chan<- interface{}
-}
diff --git a/internal/io/http/client.go b/internal/io/http/client.go
index 5288c8af..edbfb286 100644
--- a/internal/io/http/client.go
+++ b/internal/io/http/client.go
@@ -15,7 +15,9 @@
 package http
 
 import (
+	"crypto/md5"
 	"crypto/tls"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -64,12 +66,13 @@ type RefreshTokenConf struct {
 }
 
 type RawConf struct {
-	Url      string            `json:"url"`
-	Method   string            `json:"method"`
-	Body     string            `json:"body"`
-	BodyType string            `json:"bodyType"`
-	Headers  map[string]string `json:"headers"`
-	Timeout  int               `json:"timeout"`
+	Url         string            `json:"url"`
+	Method      string            `json:"method"`
+	Body        string            `json:"body"`
+	BodyType    string            `json:"bodyType"`
+	Headers     map[string]string `json:"headers"`
+	Timeout     int               `json:"timeout"`
+	Incremental bool              `json:"incremental"`
 
 	OAuth map[string]map[string]interface{} `json:"oauth"`
 	// Could be code or body
@@ -219,7 +222,7 @@ func (cc *ClientConf) auth(ctx api.StreamContext) error {
 	if err != nil {
 		return err
 	}
-	tokens, _, err := cc.parseResponse(ctx, resp)
+	tokens, _, err := cc.parseResponse(ctx, resp, "")
 	if err != nil {
 		return err
 	}
@@ -271,7 +274,7 @@ func (cc *ClientConf) refresh(ctx api.StreamContext) error {
 		if err != nil {
 			return fmt.Errorf("fail to get refresh token: %v", err)
 		}
-		nt, _, err := cc.parseResponse(ctx, resp)
+		nt, _, err := cc.parseResponse(ctx, resp, "")
 		if err != nil {
 			return fmt.Errorf("Cannot parse refresh token response to json: %v", err)
 		}
@@ -310,55 +313,63 @@ func (cc *ClientConf) responseBodyDecompress(ctx api.StreamContext, resp *http.R
 }
 
 // parse the response status. For rest sink, it will not return the body by default if not need to debug
-func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response) ([]map[string]interface{}, []byte, error) {
+func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response, lastMD5 string) ([]map[string]interface{}, string, error) {
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
-		return nil, nil, fmt.Errorf("%s: %d", CODE_ERR, resp.StatusCode)
+		return nil, "", fmt.Errorf("%s: %d", CODE_ERR, resp.StatusCode)
 	}
 
 	c, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, nil, fmt.Errorf("%s: %v", BODY_ERR, err)
+		return nil, "", fmt.Errorf("%s: %v", BODY_ERR, err)
 	}
 
 	defer func() {
 		resp.Body.Close()
 	}()
 
+	newMD5 := ""
+	if cc.config.Incremental {
+		newMD5 = getMD5Hash(c)
+		if newMD5 == lastMD5 {
+			return nil, newMD5, nil
+		}
+	}
+
 	switch cc.config.ResponseType {
 	case "code":
 		if cc.config.Compression != "" {
 			if c, err = cc.responseBodyDecompress(ctx, resp, c); err != nil {
-				return nil, nil, fmt.Errorf("try to decompress payload failed, %w", err)
+				return nil, "", fmt.Errorf("try to decompress payload failed, %w", err)
 			}
 		}
 		m, e := decode(c)
 		if e != nil {
-			return nil, c, fmt.Errorf("%s: decode fail for %v", BODY_ERR, e)
+			return nil, "", fmt.Errorf("%s: decode fail for %v", BODY_ERR, e)
 		}
-		return m, c, e
+		return m, newMD5, e
 	case "body":
 		if cc.config.Compression != "" {
 			if c, err = cc.responseBodyDecompress(ctx, resp, c); err != nil {
-				return nil, nil, fmt.Errorf("try to decompress payload failed, %w", err)
+				return nil, "", fmt.Errorf("try to decompress payload failed, %w", err)
 			}
 		}
 		payloads, err := decode(c)
 		if err != nil {
-			return nil, c, fmt.Errorf("%s: decode fail for %v", BODY_ERR, err)
+			return nil, "", fmt.Errorf("%s: decode fail for %v", BODY_ERR, err)
 		}
 		for _, payload := range payloads {
 			ro := &bodyResp{}
 			err = cast.MapToStruct(payload, ro)
 			if err != nil {
-				return nil, c, fmt.Errorf("%s: decode fail for %v", BODY_ERR, err)
+				return nil, "", fmt.Errorf("%s: decode fail for %v", BODY_ERR, err)
 			}
 			if ro.Code < 200 || ro.Code > 299 {
-				return nil, c, fmt.Errorf("%s: %d", CODE_ERR, ro.Code)
+				return nil, "", fmt.Errorf("%s: %d", CODE_ERR, ro.Code)
 			}
 		}
-		return payloads, c, nil
+		return payloads, newMD5, nil
 	default:
-		return nil, c, fmt.Errorf("%s: unsupported response type %s", BODY_ERR, cc.config.ResponseType)
+		return nil, "", fmt.Errorf("%s: unsupported response type %s", BODY_ERR, cc.config.ResponseType)
 	}
 }
 
@@ -390,3 +401,8 @@ func decode(data []byte) ([]map[string]interface{}, error) {
 	}
 	return nil, fmt.Errorf("only map[string]interface{} and []map[string]interface{} is supported")
 }
+
+func getMD5Hash(text []byte) string {
+	hash := md5.Sum(text)
+	return hex.EncodeToString(hash[:])
+}
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index 93b6b755..a4111232 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -24,6 +24,7 @@ import (
 
 type HttpPullSource struct {
 	*ClientConf
+	lastMD5 string
 }
 
 func (hps *HttpPullSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest) {
@@ -67,10 +68,11 @@ func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, erro
 	if err != nil {
 		return nil, err
 	}
-	results, _, err := hps.parseResponse(ctx, resp)
+	results, newMD5, err := hps.parseResponse(ctx, resp, hps.lastMD5)
 	if err != nil {
 		return nil, err
 	}
+	hps.lastMD5 = newMD5
 	return results, nil
 }
 
diff --git a/internal/plugin/js/exporter.go b/internal/plugin/js/exporter.go
new file mode 100644
index 00000000..7ed42c8a
--- /dev/null
+++ b/internal/plugin/js/exporter.go
@@ -0,0 +1,87 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"encoding/json"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+)
+
+// Exporter is used to export and import the JavaScript functions
+// The functions are stored in the key-value store only. So import and export are just to read and write the key-value store
+
+// Import the JavaScript functions from the map. This is usually called after reset to override all settings
+func (m *Manager) Import(scripts map[string]string) map[string]string {
+	errMap := map[string]string{}
+	_ = m.importStatusDb.Clean()
+	for k, v := range scripts {
+		err := m.UpsertByJson(k, v)
+		if err != nil {
+			_ = m.importStatusDb.Set(k, err.Error())
+			errMap[k] = err.Error()
+		}
+	}
+	return errMap
+}
+
+func (m *Manager) PartialImport(scripts map[string]string) map[string]string {
+	errMap := map[string]string{}
+	for k, v := range scripts {
+		err := m.UpsertByJson(k, v)
+		if err != nil {
+			errMap[k] = err.Error()
+		}
+	}
+	return errMap
+}
+
+func (m *Manager) Export() map[string]string {
+	all, err := m.db.Keys()
+	if err != nil {
+		conf.Log.Errorf("Fail to export the JavaScript function manager: %v", err)
+		return nil
+	}
+	result := make(map[string]string, len(all))
+	for _, k := range all {
+		s, err := m.GetScript(k)
+		if err != nil {
+			conf.Log.Errorf("Fail to export the JavaScript function %s: %v", k, err)
+			continue
+		}
+		sj, err := json.Marshal(s)
+		if err != nil {
+			conf.Log.Errorf("Fail to marshal the JavaScript function %s: %v", k, err)
+			continue
+		}
+		result[k] = string(sj)
+	}
+	return result
+}
+
+func (m *Manager) Status() map[string]string {
+	all, err := m.importStatusDb.All()
+	if err != nil {
+		return nil
+	}
+	return all
+}
+
+func (m *Manager) Reset() {
+	err := m.db.Clean()
+	if err != nil {
+		conf.Log.Errorf("Fail to reset the JavaScript function manager: %v", err)
+	}
+}
diff --git a/internal/plugin/js/factory.go b/internal/plugin/js/factory.go
new file mode 100644
index 00000000..856ac07d
--- /dev/null
+++ b/internal/plugin/js/factory.go
@@ -0,0 +1,46 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func (m *Manager) Function(name string) (api.Function, error) {
+	f, err := NewJSFunc(name)
+	if err != nil {
+		return nil, err
+	}
+	return f, nil
+}
+
+func (m *Manager) HasFunctionSet(_ string) bool {
+	return false
+}
+
+func (m *Manager) FunctionPluginInfo(funcName string) (plugin.EXTENSION_TYPE, string, string) {
+	_, ok := m.ConvName(funcName)
+	if !ok {
+		return plugin.NONE_EXTENSION, "", ""
+	} else {
+		return plugin.JS_EXTENSION, "", ""
+	}
+}
+
+func (m *Manager) ConvName(n string) (string, bool) {
+	_, err := m.GetScript(n)
+	return n, err == nil
+}
diff --git a/internal/plugin/js/function.go b/internal/plugin/js/function.go
new file mode 100644
index 00000000..bdb766d1
--- /dev/null
+++ b/internal/plugin/js/function.go
@@ -0,0 +1,97 @@
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/dop251/goja"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+)
+
+// JSFunc is stateful
+// Each instance has its own vm
+type JSFunc struct {
+	vm     *goja.Runtime
+	jsfunc goja.Callable
+	isAgg  bool
+	// state, use this to avoid creating new array each time
+	args []goja.Value
+}
+
+func NewJSFunc(symbolName string) (*JSFunc, error) {
+	s, err := GetManager().GetScript(symbolName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get script for %s: %v", symbolName, err)
+	}
+	vm := goja.New()
+	// Get the text from the symbol table
+	_, _ = vm.RunString(s.Script)
+	// Should not happen, already verify when install
+	//if err != nil {
+	//	return nil, fmt.Errorf("failed to interpret script: %v", err)
+	//}
+	exec, _ := goja.AssertFunction(vm.Get(symbolName))
+	// Should not happen, already verify when install
+	//if !ok {
+	//	return nil, fmt.Errorf("cannot find function \"%s\" in script", symbolName)
+	//}
+	return &JSFunc{
+		vm:     vm,
+		jsfunc: exec,
+		isAgg:  s.IsAgg,
+	}, nil
+}
+
+func (f *JSFunc) Validate(_ []interface{}) error {
+	return nil
+}
+
+func (f *JSFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+	ctx.GetLogger().Debugf("running js func with args %+v", args)
+	if len(args) != len(f.args) {
+		f.args = make([]goja.Value, len(args))
+	}
+	for i, arg := range args {
+		f.args[i] = f.vm.ToValue(arg)
+	}
+	val, err := f.jsfunc(goja.Undefined(), f.args...)
+	if err != nil {
+		ctx.GetLogger().Errorf("failed to execute script: %v", err)
+		return err, false
+	} else {
+		result := val.Export()
+		switch t := result.(type) {
+		case float64:
+			if math.IsNaN(t) {
+				return fmt.Errorf("result is NaN"), false
+			}
+			if math.IsInf(t, 0) {
+				return fmt.Errorf("result is Inf"), false
+			}
+		}
+		return result, true
+	}
+}
+
+func (f *JSFunc) IsAggregate() bool {
+	return f.isAgg
+}
+
+func (f *JSFunc) Close() error {
+	return nil
+}
diff --git a/internal/plugin/js/manager.go b/internal/plugin/js/manager.go
new file mode 100644
index 00000000..3c8b6191
--- /dev/null
+++ b/internal/plugin/js/manager.go
@@ -0,0 +1,119 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/dop251/goja"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/pkg/kv"
+)
+
+var manager *Manager
+
+func GetManager() *Manager {
+	return manager
+}
+
+type Manager struct {
+	db             kv.KeyValue
+	importStatusDb kv.KeyValue
+}
+
+type Script struct {
+	Id     string `json:"id"`
+	Desc   string `json:"description"`
+	Script string `json:"script"`
+	IsAgg  bool   `json:"isAgg"`
+}
+
+// InitManager initialize the manager, only called once by the server
+func InitManager() error {
+	db, err := store.GetKV("script")
+	if err != nil {
+		return fmt.Errorf("can not initialize store for the JavaScript function manager at path 'script': %v", err)
+	}
+	importStatusDb, err := store.GetKV("scriptInstallStatus")
+	if err != nil {
+		return fmt.Errorf("can not initialize store for the JavaScript function manager at path 'scriptInstallStatus': %v", err)
+	}
+
+	manager = &Manager{
+		db:             db,
+		importStatusDb: importStatusDb,
+	}
+	return nil
+}
+
+func (m *Manager) UpsertByJson(k string, v string) error {
+	s := &Script{Id: k}
+	err := json.Unmarshal([]byte(v), s)
+	if err != nil {
+		return fmt.Errorf("fail to unmarshal the script %s: %v", k, err)
+	}
+	if s.Id != k {
+		return fmt.Errorf("the script id %s does not match the key %s", s.Id, k)
+	}
+	return m.Update(s)
+}
+
+func (m *Manager) Create(script *Script) error {
+	err := validate(script)
+	if err != nil {
+		return err
+	}
+	return m.db.Setnx(script.Id, script)
+}
+
+func validate(script *Script) error {
+	vm := goja.New()
+	_, err := vm.RunString(script.Script)
+	if err != nil {
+		return fmt.Errorf("failed to interprete script: %v", err)
+	}
+	_, ok := goja.AssertFunction(vm.Get(script.Id))
+	if !ok {
+		return fmt.Errorf("cannot find function \"%s\" in script", script.Id)
+	}
+	return nil
+}
+
+func (m *Manager) GetScript(id string) (*Script, error) {
+	result := &Script{}
+	ok, err := m.db.Get(id, result)
+	if !ok && err == nil {
+		return nil, fmt.Errorf("not found")
+	}
+	return result, err
+}
+
+func (m *Manager) List() ([]string, error) {
+	return m.db.Keys()
+}
+
+func (m *Manager) Update(script *Script) error {
+	err := validate(script)
+	if err != nil {
+		return err
+	}
+	return m.db.Set(script.Id, script)
+}
+
+func (m *Manager) Delete(id string) error {
+	return m.db.Delete(id)
+}
diff --git a/internal/plugin/plugin.go b/internal/plugin/plugin.go
index 2a2ccd04..f3be9c6f 100644
--- a/internal/plugin/plugin.go
+++ b/internal/plugin/plugin.go
@@ -23,7 +23,6 @@ const (
 	SINK
 	FUNCTION
 	PORTABLE
-	WASM
 )
 
 var PluginTypes = []string{"sources", "sinks", "functions", "portable", "wasm"}
@@ -33,7 +32,6 @@ var PluginTypeMap = map[string]PluginType{
 	"sinks":     SINK,
 	"functions": FUNCTION,
 	"portable":  PORTABLE,
-	"wasm":      WASM,
 }
 
 type Plugin interface {
@@ -84,8 +82,6 @@ func NewPluginByType(t PluginType) Plugin {
 	switch t {
 	case FUNCTION:
 		return &FuncPlugin{}
-	case WASM:
-		return &FuncPlugin{}
 	default:
 		return &IOPlugin{}
 	}
@@ -109,6 +105,5 @@ const (
 	NATIVE_EXTENSION
 	PORTABLE_EXTENSION
 	SERVICE_EXTENSION
-	WASM_EXTENSION
 	JS_EXTENSION
 )
diff --git a/internal/server/script_init.go b/internal/server/script_init.go
new file mode 100644
index 00000000..2e753c50
--- /dev/null
+++ b/internal/server/script_init.go
@@ -0,0 +1,117 @@
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build script || full
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/gorilla/mux"
+
+	"github.com/lf-edge/ekuiper/v2/internal/binder"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/js"
+)
+
+func init() {
+	components["script"] = scriptComp{}
+}
+
+type scriptComp struct{}
+
+func (p scriptComp) register() {
+	err := js.InitManager()
+	if err != nil {
+		panic(err)
+	}
+	entries = append(entries, binder.FactoryEntry{Name: "javascript function", Factory: js.GetManager(), Weight: 7})
+}
+
+func (p scriptComp) rest(r *mux.Router) {
+	r.HandleFunc("/udf/javascript", jsfuncsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/udf/javascript/{id}", jsfuncHandler).Methods(http.MethodGet, http.MethodDelete, http.MethodPut)
+}
+
+func (p scriptComp) exporter() ConfManager {
+	return js.GetManager()
+}
+
+func jsfuncsHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	switch r.Method {
+	case http.MethodGet:
+		content, err := js.GetManager().List()
+		if err != nil {
+			handleError(w, err, "jsfuncs list command error", logger)
+			return
+		}
+		jsonResponse(content, w, logger)
+	case http.MethodPost:
+		sd := &js.Script{}
+		err := json.NewDecoder(r.Body).Decode(sd)
+		// Problems decoding
+		if err != nil {
+			handleError(w, err, "Invalid body: Error decoding the new javascript function json", logger)
+			return
+		}
+		err = js.GetManager().Create(sd)
+		if err != nil {
+			handleError(w, err, "javascript function create command error", logger)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		fmt.Fprintf(w, "javascript function %s is created", sd.Id)
+	}
+}
+
+func jsfuncHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	name := vars["id"]
+	switch r.Method {
+	case http.MethodDelete:
+		err := js.GetManager().Delete(name)
+		if err != nil {
+			handleError(w, err, fmt.Sprintf("delete javascript function %s error", name), logger)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "javascript function %s is deleted", name)
+	case http.MethodGet:
+		j, err := js.GetManager().GetScript(name)
+		if err != nil {
+			handleError(w, err, fmt.Sprintf("describe javascript function %s error", name), logger)
+			return
+		}
+		jsonResponse(j, w, logger)
+	case http.MethodPut:
+		sd := &js.Script{}
+		err := json.NewDecoder(r.Body).Decode(sd)
+		// Problems decoding
+		if err != nil {
+			handleError(w, err, "Invalid body: Error decoding the javascript function json", logger)
+			return
+		}
+		err = js.GetManager().Update(sd)
+		if err != nil {
+			handleError(w, err, "javascript function update command error", logger)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "javascript function %s is updated", sd.Id)
+	}
+}
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index 8a6e9b92..80b46b01 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -87,12 +87,15 @@ func NewLookupNode(ctx api.StreamContext, name string, isBytesLookup bool, field
 		if (sc.PayloadFormat == "" && sc.PayloadField != "") || (sc.PayloadFormat != "" && sc.PayloadField == "") {
 			return nil, fmt.Errorf("payloadFormat and payloadField must set together")
 		}
-		sch := make(map[string]*ast.JsonStreamField, len(fields))
-		for _, field := range fields {
-			sch[field] = nil
-		}
-		if sc.PayloadField != "" {
-			sch[sc.PayloadField] = nil
+		var sch map[string]*ast.JsonStreamField
+		if len(fields) > 0 {
+			sch = make(map[string]*ast.JsonStreamField, len(fields))
+			for _, field := range fields {
+				sch[field] = nil
+			}
+			if sc.PayloadField != "" {
+				sch[sc.PayloadField] = nil
+			}
 		}
 
 		decoder, err := converter.GetOrCreateConverter(ctx, srcOptions.FORMAT, srcOptions.SCHEMAID, sch, props)
diff --git a/pkg/nng/sock.go b/pkg/nng/sock.go
index 1eae58f5..3c311d98 100644
--- a/pkg/nng/sock.go
+++ b/pkg/nng/sock.go
@@ -146,7 +146,7 @@ func (s *Sock) Send(ctx api.StreamContext, data []byte) error {
 	if s.Socket != nil && s.connected.Load() {
 		return s.Socket.Send(data)
 	}
-	return errorx.NewIOErr(`neuron connection is not established`)
+	return errorx.NewIOErr(`nng connection is not established`)
 }
 
 var _ modules.Connection = &Sock{}


Test patch:
diff --git a/internal/io/sql/client/client.go b/extension/sql/client/client.go
similarity index 100%
rename from internal/io/sql/client/client.go
rename to extension/sql/client/client.go
diff --git a/internal/io/sql/client/client_test.go b/extension/sql/client/client_test.go
similarity index 96%
rename from internal/io/sql/client/client_test.go
rename to extension/sql/client/client_test.go
index e69cf981..1a0cd272 100644
--- a/internal/io/sql/client/client_test.go
+++ b/extension/sql/client/client_test.go
@@ -20,7 +20,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
diff --git a/internal/io/sql/sink_test.go b/extension/sql/sink_test.go
similarity index 99%
rename from internal/io/sql/sink_test.go
rename to extension/sql/sink_test.go
index 07957d80..56d51a1a 100644
--- a/internal/io/sql/sink_test.go
+++ b/extension/sql/sink_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/io/sql/testx"
+
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
diff --git a/internal/io/sql/source_test.go b/extension/sql/source_test.go
similarity index 98%
rename from internal/io/sql/source_test.go
rename to extension/sql/source_test.go
index e69d3596..c6d55da1 100644
--- a/internal/io/sql/source_test.go
+++ b/extension/sql/source_test.go
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build full
-// +build full
-
 package sql
 
 import (
@@ -27,8 +24,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extension/sql/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
diff --git a/internal/io/http/httppull_source_test.go b/internal/io/http/httppull_source_test.go
index e76cedaa..afaa663f 100644
--- a/internal/io/http/httppull_source_test.go
+++ b/internal/io/http/httppull_source_test.go
@@ -32,6 +32,15 @@ type Response struct {
 	Code    int    `json:"code"`
 }
 
+func handleCodeErr(w http.ResponseWriter, r *http.Request) {
+	resp := Response{
+		Message: "Err",
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusBadRequest)
+	json.NewEncoder(w).Encode(resp)
+}
+
 func handleGet(w http.ResponseWriter, r *http.Request) {
 	resp := Response{
 		Message: "Hello, GET!",
@@ -86,6 +95,7 @@ func createServer() *httptest.Server {
 	router.HandleFunc("/get", handleGet)
 	router.HandleFunc("/post", handlePost)
 	router.HandleFunc("/err", handleErr)
+	router.HandleFunc("/codeErr", handleCodeErr)
 	router.HandleFunc("/auth", handleAuth)
 	router.HandleFunc("/refresh", handleRefresh)
 	server := httptest.NewServer(router)
@@ -152,4 +162,49 @@ func TestHttpPullSource(t *testing.T) {
 	require.Error(t, <-errCh)
 	close(errCh)
 	source.Close(ctx)
+
+	source = &HttpPullSource{}
+	require.NoError(t, source.Provision(ctx, map[string]any{
+		"url":          server.URL,
+		"datasource":   "/codeErr",
+		"method":       "post",
+		"responseType": "code",
+	}))
+	require.NoError(t, source.Connect(ctx))
+	errCh = make(chan error, 1)
+	source.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {}, func(ctx api.StreamContext, err error) { errCh <- err })
+	require.Error(t, <-errCh)
+	close(errCh)
+	source.Close(ctx)
+}
+
+func TestSourceIncremental(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	source := &HttpPullSource{}
+	require.NoError(t, source.Provision(ctx, map[string]any{
+		"url":         server.URL,
+		"datasource":  "/get",
+		"method":      "get",
+		"incremental": true,
+	}))
+	require.NoError(t, source.Connect(ctx))
+	dataCh := make(chan any, 1)
+	source.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		dataCh <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Equal(t, []map[string]interface{}{
+		{
+			"message": "Hello, GET!",
+			"code":    float64(200),
+		},
+	}, <-dataCh)
+
+	source.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		dataCh <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Nil(t, <-dataCh)
 }
diff --git a/internal/plugin/js/export_test.go b/internal/plugin/js/export_test.go
new file mode 100644
index 00000000..65db4ab6
--- /dev/null
+++ b/internal/plugin/js/export_test.go
@@ -0,0 +1,160 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFullLifecycle(t *testing.T) {
+	m := GetManager()
+	m.Reset()
+	// Create scripts
+	script := &Script{
+		Id:     "testScript",
+		Desc:   "Test script",
+		Script: "function testScript() { return 'Hello, World!'; }",
+		IsAgg:  false,
+	}
+	err := m.Create(script)
+	assert.NoError(t, err)
+	script = &Script{
+		Id:     "area",
+		Desc:   "func for area",
+		Script: "function area(x, y) { return x*y; }",
+		IsAgg:  false,
+	}
+	err = m.Create(script)
+	assert.NoError(t, err)
+	expected := map[string]string{
+		"area":       "{\"id\":\"area\",\"description\":\"func for area\",\"script\":\"function area(x, y) { return x*y; }\",\"isAgg\":false}",
+		"testScript": "{\"id\":\"testScript\",\"description\":\"Test script\",\"script\":\"function testScript() { return 'Hello, World!'; }\",\"isAgg\":false}",
+	}
+	// Export scripts
+	scripts := m.Export()
+	assert.Equal(t, expected, scripts)
+	// Partial import with error
+	invalidSet := map[string]string{
+		"area2":      "{\"id\":\"area2\",\"description\":\"func for area2\",\"script\":\"function area2(x, y) { return x*y; }\",\"isAgg\":false}",
+		"testScript": "{\"id\":\"testScript\",\"description\":\"Test script\",\"script\":\"function testScript() { return 'Hello, new!'; }\",\"isAgg\":false}",
+		"invalid":    "{\"id\":\"novalid\",\"description\":\"invalid script\",\"script\":\"function novalid() { return 'Hello, invalid!'; }\",\"isAgg\":false}",
+	}
+	errMap := m.PartialImport(invalidSet)
+	assert.Equal(t, map[string]string{"invalid": "the script id novalid does not match the key invalid"}, errMap)
+	newExpected := map[string]string{
+		"area":       "{\"id\":\"area\",\"description\":\"func for area\",\"script\":\"function area(x, y) { return x*y; }\",\"isAgg\":false}",
+		"testScript": "{\"id\":\"testScript\",\"description\":\"Test script\",\"script\":\"function testScript() { return 'Hello, new!'; }\",\"isAgg\":false}",
+		"area2":      "{\"id\":\"area2\",\"description\":\"func for area2\",\"script\":\"function area2(x, y) { return x*y; }\",\"isAgg\":false}",
+	}
+	scripts = m.Export()
+	assert.Equal(t, newExpected, scripts)
+	// Import scripts
+	m.Reset()
+	errMap = m.Import(expected)
+	assert.Empty(t, errMap)
+	status := m.Status()
+	assert.Empty(t, status)
+	scripts = m.Export()
+	assert.Equal(t, expected, scripts)
+	// Import with error
+	m.Reset()
+	scripts = m.Export()
+	assert.Empty(t, scripts)
+	errMap = m.Import(invalidSet)
+	assert.Equal(t, map[string]string{"invalid": "the script id novalid does not match the key invalid"}, errMap)
+	status = m.Status()
+	assert.Equal(t, map[string]string{"invalid": "the script id novalid does not match the key invalid"}, status)
+	newExpected2 := map[string]string{
+		"testScript": "{\"id\":\"testScript\",\"description\":\"Test script\",\"script\":\"function testScript() { return 'Hello, new!'; }\",\"isAgg\":false}",
+		"area2":      "{\"id\":\"area2\",\"description\":\"func for area2\",\"script\":\"function area2(x, y) { return x*y; }\",\"isAgg\":false}",
+	}
+	scripts = m.Export()
+	assert.Equal(t, newExpected2, scripts)
+	// Delete scripts
+	m.Reset()
+	scripts = m.Export()
+	assert.Empty(t, scripts)
+}
+
+func TestDBInvalid(t *testing.T) {
+	// mock db error
+	m := GetManager()
+	oldDb := m.db
+	m.db = &mockInvalidDB{}
+	defer func() {
+		m.db = oldDb
+	}()
+	m.Reset()
+	errMap := m.Import(map[string]string{"test": "{\"id\":\"test\",\"description\":\"Test script\",\"script\":\"function test() { return 'Hello, World!'; }\",\"isAgg\":false}"})
+	assert.Equal(t, map[string]string{"test": "db is nil"}, errMap)
+	errMap = m.PartialImport(map[string]string{"test": "{\"id\":\"test\",\"description\":\"Test script\",\"script\":\"function test() { return 'Hello, World!'; }\",\"isAgg\":false}"})
+	assert.Equal(t, map[string]string{"test": "db is nil"}, errMap)
+	scripts := m.Export()
+	assert.Empty(t, scripts)
+	status := m.Status()
+	assert.Equal(t, map[string]string{"test": "db is nil"}, status)
+	oldStatusDb := m.importStatusDb
+	m.importStatusDb = &mockInvalidDB{}
+	defer func() {
+		m.importStatusDb = oldStatusDb
+	}()
+	status = m.Status()
+	assert.Empty(t, status)
+}
+
+type mockInvalidDB struct{}
+
+func (m mockInvalidDB) Setnx(key string, value interface{}) error {
+	return fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Set(key string, value interface{}) error {
+	return fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Get(key string, val interface{}) (bool, error) {
+	return false, fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) GetKeyedState(key string) (interface{}, error) {
+	return false, fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) SetKeyedState(key string, value interface{}) error {
+	return fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Delete(key string) error {
+	return fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Keys() (keys []string, err error) {
+	return nil, fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) All() (all map[string]string, err error) {
+	return nil, fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Clean() error {
+	return fmt.Errorf("db is nil")
+}
+
+func (m mockInvalidDB) Drop() error {
+	return fmt.Errorf("db is nil")
+}
diff --git a/internal/plugin/js/factory_test.go b/internal/plugin/js/factory_test.go
new file mode 100644
index 00000000..a96e925e
--- /dev/null
+++ b/internal/plugin/js/factory_test.go
@@ -0,0 +1,59 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func TestFactory(t *testing.T) {
+	script := &Script{
+		Id:     "existingFunction",
+		Desc:   "Existing function",
+		Script: "function existingFunction() { return 'Hello, World!'; }",
+		IsAgg:  false,
+	}
+	err := GetManager().Create(script)
+	assert.NoError(t, err)
+	defer func() {
+		err := GetManager().Delete("existingFunction")
+		assert.NoError(t, err)
+	}()
+	_, err = GetManager().Function("existingFunction")
+	assert.NoError(t, err)
+	r := GetManager().HasFunctionSet("existingFunction")
+	assert.False(t, r)
+	ft, _, _ := GetManager().FunctionPluginInfo("existingFunction")
+	assert.Equal(t, plugin.JS_EXTENSION, ft)
+	n, ok := GetManager().ConvName("existingFunction")
+	assert.True(t, ok)
+	assert.Equal(t, "existingFunction", n)
+}
+
+func TestNotExistFactory(t *testing.T) {
+	_, err := GetManager().Function("existingFunction")
+	assert.Error(t, err)
+	r := GetManager().HasFunctionSet("existingFunction")
+	assert.False(t, r)
+	ft, _, _ := GetManager().FunctionPluginInfo("nonExistentFunction")
+	assert.Equal(t, plugin.NONE_EXTENSION, ft)
+	n, ok := GetManager().ConvName("existingFunction")
+	assert.False(t, ok)
+	assert.Equal(t, "existingFunction", n)
+}
diff --git a/internal/plugin/js/function_test.go b/internal/plugin/js/function_test.go
new file mode 100644
index 00000000..fe920224
--- /dev/null
+++ b/internal/plugin/js/function_test.go
@@ -0,0 +1,231 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	kctx "github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
+)
+
+func TestScalarFuncHappyPath(t *testing.T) {
+	script := &Script{
+		Id:     "area",
+		Desc:   "Test script",
+		Script: "function area(x, y) { return x * y; }",
+		IsAgg:  false,
+	}
+	err := GetManager().Create(script)
+	assert.NoError(t, err)
+	defer func() {
+		err := GetManager().Delete("area")
+		assert.NoError(t, err)
+	}()
+	ff, err := NewJSFunc("area")
+	assert.NoError(t, err)
+	err = ff.Validate([]interface{}{})
+	assert.NoError(t, err)
+	isAgg := ff.IsAggregate()
+	assert.Equal(t, false, isAgg)
+
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", def.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+
+	tests := []struct {
+		name string
+		args []any
+		want any
+	}{
+		{
+			name: "normal",
+			args: []any{2, 3},
+			want: int64(6),
+		},
+		{
+			name: "more args",
+			args: []any{2, 3, 4},
+			want: int64(6),
+		},
+		{
+			name: "less args",
+			args: []any{2},
+			want: errors.New("result is NaN"),
+		},
+		{
+			name: "float",
+			args: []any{2.0, 3.0},
+			want: int64(6),
+		},
+		{
+			name: "float2",
+			args: []any{2.5, 3.5},
+			want: 8.75,
+		},
+		{
+			name: "string",
+			args: []any{"2", "3"},
+			want: int64(6),
+		},
+		{
+			name: "string float",
+			args: []any{"2.0", 3.0},
+			want: int64(6),
+		},
+		{
+			name: "invalid string",
+			args: []any{"myname", "hello"},
+			want: errors.New("result is NaN"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, _ := ff.Exec(tt.args, fctx)
+			assert.Equal(t, tt.want, result)
+		})
+	}
+	err = ff.Close()
+	assert.NoError(t, err)
+}
+
+func TestNonExistentScript(t *testing.T) {
+	_, err := NewJSFunc("nonExistentScript")
+	assert.NotNil(t, err)
+}
+
+func TestAggFuncHappyPath(t *testing.T) {
+	script := &Script{
+		Id:     "areas",
+		Desc:   "Test script",
+		Script: "function areas(x, y) { if(x.length !== y.length) {throw 'length of x and y should be the same'} let result = 0; for (let i = 0; i < x.length; i++) { result+=(x[i] * y[i]); } return result; }",
+		IsAgg:  true,
+	}
+	err := GetManager().Create(script)
+	assert.NoError(t, err)
+	defer func() {
+		err := GetManager().Delete("areas")
+		assert.NoError(t, err)
+	}()
+	ff, err := NewJSFunc("areas")
+	assert.NoError(t, err)
+	err = ff.Validate([]interface{}{})
+	assert.NoError(t, err)
+	isAgg := ff.IsAggregate()
+	assert.Equal(t, true, isAgg)
+
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", def.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+
+	tests := []struct {
+		name string
+		args []any
+		want any
+	}{
+		{
+			name: "normal",
+			args: []any{[]any{2, 3}, []any{3, 4}},
+			want: int64(18),
+		},
+		{
+			name: "partial error",
+			args: []any{[]any{2, 3}, []any{"ada", 4}},
+			want: errors.New("result is NaN"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, _ := ff.Exec(tt.args, fctx)
+			assert.Equal(t, tt.want, result)
+		})
+	}
+	err = ff.Close()
+	assert.NoError(t, err)
+}
+
+func TestScalarFuncComplexType(t *testing.T) {
+	script := &Script{
+		Id:     "area2",
+		Desc:   "Test script",
+		Script: "function area2(msg) { return msg.x * msg.y; }",
+		IsAgg:  false,
+	}
+	err := GetManager().Create(script)
+	assert.NoError(t, err)
+	defer func() {
+		err := GetManager().Delete("area2")
+		assert.NoError(t, err)
+	}()
+	ff, err := NewJSFunc("area2")
+	assert.NoError(t, err)
+	err = ff.Validate([]interface{}{})
+	assert.NoError(t, err)
+	isAgg := ff.IsAggregate()
+	assert.Equal(t, false, isAgg)
+
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", def.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+
+	tests := []struct {
+		name string
+		args []any
+		want any
+	}{
+		{
+			name: "normal",
+			args: []any{map[string]any{
+				"x": 2,
+				"y": 3,
+			}},
+			want: int64(6),
+		},
+		{
+			name: "more args",
+			args: []any{map[string]any{
+				"x": 2,
+				"y": 3,
+				"z": 4,
+			}},
+			want: int64(6),
+		},
+		{
+			name: "invalid string",
+			args: []any{map[string]any{
+				"x": 2,
+				"y": "ddd",
+				"z": 4,
+			}},
+			want: errors.New("result is NaN"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, _ := ff.Exec(tt.args, fctx)
+			assert.Equal(t, tt.want, result)
+		})
+	}
+	err = ff.Close()
+	assert.NoError(t, err)
+}
diff --git a/internal/plugin/js/manager_test.go b/internal/plugin/js/manager_test.go
new file mode 100644
index 00000000..9bae7c6e
--- /dev/null
+++ b/internal/plugin/js/manager_test.go
@@ -0,0 +1,116 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package js
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+)
+
+func init() {
+	testx.InitEnv("js")
+	// Wait for other db tests to finish to avoid db lock
+	for i := 0; i < 10; i++ {
+		if err := InitManager(); err != nil {
+			time.Sleep(10 * time.Millisecond)
+		} else {
+			break
+		}
+	}
+}
+
+func TestCorrectLC(t *testing.T) {
+	script := &Script{
+		Id:     "testScript",
+		Desc:   "Test script",
+		Script: "function testScript() { return 'Hello, World!'; }",
+		IsAgg:  false,
+	}
+	err := GetManager().Create(script)
+	assert.Nil(t, err)
+	result, err := GetManager().GetScript("testScript")
+	assert.Nil(t, err)
+	assert.Equal(t, script, result)
+	keys, err := GetManager().List()
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"testScript"}, keys)
+	err = GetManager().Delete("testScript")
+	assert.Nil(t, err)
+}
+
+func TestCreateScriptWithInvalidFunction(t *testing.T) {
+	script := &Script{
+		Id:     "invalidScript",
+		Desc:   "Invalid script",
+		Script: "function invalidScript() { return 'Hello, World!';",
+		IsAgg:  false,
+	}
+	err := GetManager().Create(script)
+	assert.NotNil(t, err)
+
+	err = GetManager().UpsertByJson("invalidScript", "{\"id\":\"invalidScript\",\"description\":\"Invalid script\",\"script\":\"function invalidScript() { return 'Hello, World!';\",\"isAgg\":false}")
+	assert.Error(t, err)
+
+	script = &Script{
+		Id:     "invalidName",
+		Desc:   "Invalid script",
+		Script: "function area() { return 'Hello, World!';}",
+		IsAgg:  false,
+	}
+	err = GetManager().Create(script)
+	assert.NotNil(t, err)
+}
+
+func TestGetScriptNonExistent(t *testing.T) {
+	_, err := GetManager().GetScript("nonExistentScript")
+	assert.NotNil(t, err)
+}
+
+func TestUpdateScript(t *testing.T) {
+	script := &Script{
+		Id:     "testScript",
+		Desc:   "Test script",
+		Script: "function testScript(x, y) { return x*y; }",
+		IsAgg:  false,
+	}
+	GetManager().Create(script)
+	script.Script = "function testScript() { return x + y; }"
+	err := GetManager().Update(script)
+	assert.Nil(t, err)
+	// Cannot create existing script
+	err = GetManager().Create(script)
+	assert.NotNil(t, err)
+	newScript := &Script{
+		Id:     "testScript",
+		Desc:   "Test script",
+		Script: "function wrongName() { return x + y; }",
+	}
+	err = GetManager().Update(newScript)
+	assert.NotNil(t, err)
+	result, err := GetManager().GetScript("testScript")
+	assert.Nil(t, err)
+	assert.Equal(t, script, result)
+	err = GetManager().Delete("testScript")
+	assert.Nil(t, err)
+}
+
+func TestDeleteNonExistentScript(t *testing.T) {
+	err := GetManager().Delete("nonExistentScript")
+	assert.NotNil(t, err)
+}
diff --git a/internal/server/script_init_test.go b/internal/server/script_init_test.go
new file mode 100644
index 00000000..777c5621
--- /dev/null
+++ b/internal/server/script_init_test.go
@@ -0,0 +1,117 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build script || full
+
+package server
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/suite"
+)
+
+type ScriptTestSuite struct {
+	suite.Suite
+	sc scriptComp
+	r  *mux.Router
+}
+
+func (suite *ScriptTestSuite) SetupTest() {
+	suite.sc = scriptComp{}
+	suite.r = mux.NewRouter()
+	suite.sc.register()
+	suite.sc.rest(suite.r)
+}
+
+func (suite *ScriptTestSuite) TestAPI() {
+	// create correct
+	body := `{"id": "area", "description": "function to calculate area", "script": "function area(x, y) { return x * y; }", "is_agg": false}`
+	req, _ := http.NewRequest(http.MethodPost, "/udf/javascript", bytes.NewBufferString(body))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusCreated, w.Code)
+
+	// duplicate error
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+
+	// invalid json
+	invalid := `{"id": "area, "description": "function to calculate area", "script": "function area(x, y) { return x * y; }", "is_agg": false}`
+	req, _ = http.NewRequest(http.MethodPost, "/udf/javascript", bytes.NewBufferString(invalid))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+
+	// get correct
+	req, _ = http.NewRequest(http.MethodGet, "/udf/javascript/area", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+	exp := "{\"id\":\"area\",\"description\":\"function to calculate area\",\"script\":\"function area(x, y) { return x * y; }\",\"isAgg\":false}"
+	suite.Equal(exp, w.Body.String())
+
+	// get inexist
+	req, _ = http.NewRequest(http.MethodGet, "/udf/javascript/area1", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+
+	// list
+	req, _ = http.NewRequest(http.MethodGet, "/udf/javascript", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+	suite.Equal(`["area"]`, w.Body.String())
+
+	// update
+	body2 := `{"id": "area", "description": "function to calculate area", "script": "function area(x, y) { return x + y; }", "is_agg": false}`
+	req, _ = http.NewRequest(http.MethodPut, "/udf/javascript/area", bytes.NewBufferString(body2))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+
+	// update invalid json
+	req, _ = http.NewRequest(http.MethodPut, "/udf/javascript/area", bytes.NewBufferString(invalid))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+
+	// delete
+	req, _ = http.NewRequest(http.MethodDelete, "/udf/javascript/area", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+
+	// update in-exist, support upsert
+	req, _ = http.NewRequest(http.MethodPut, "/udf/javascript/area", bytes.NewBufferString(body))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+
+	// delete
+	req, _ = http.NewRequest(http.MethodDelete, "/udf/javascript/area", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	suite.Equal(http.StatusOK, w.Code)
+}
+
+func TestScriptTestSuite(t *testing.T) {
+	suite.Run(t, new(ScriptTestSuite))
+}