From 07ebd747abd271d8ef54dd613f057d5473fc1857 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Sat, 19 Mar 2022 11:59:43 +0700
Subject: [PATCH 01/10] [Update] Define method

---
 constants.go     |  22 ++++++---
 elasticsearch.go | 118 +++++++++++------------------------------------
 go.mod           |   6 +--
 go.sum           |   9 ++++
 pull.go          |  20 ++++++++
 push.go          |   5 ++
 queue.go         |  20 ++++++++
 request.go       |  35 ++++++++++++++
 8 files changed, 136 insertions(+), 99 deletions(-)
 create mode 100644 pull.go
 create mode 100644 push.go
 create mode 100644 queue.go
 create mode 100644 request.go

diff --git a/constants.go b/constants.go
index 81055aa..f997694 100644
--- a/constants.go
+++ b/constants.go
@@ -1,13 +1,23 @@
 package elasticsearch
 
 const (
-	SubjectSyncData       = "elasticsearch/sync_data"
-	SubjectSearch         = "elasticsearch/search"
-	SubjectUpdateDocument = "elasticsearch/update_document"
-	SubjectCreateIndex    = "elasticsearch/create_index"
-	SubjectDeleteDocument = "elasticsearch/delete_document"
+	SubjectRequestProductUpsert = "elasticsearch/selly.request.product.upsert"
+	SubjectQueueProductUpsert   = "elasticsearch/selly.queue.product.upsert"
+	SubjectPullProductUpsert    = "elasticsearch/selly.pull.product.upsert"
+	SubjectRequestProductSearch = "elasticsearch/selly.request.product.search"
+
+	SubjectRequestOrderUpsert = "elasticsearch/selly.request.order.upsert"
+	SubjectQueueOrderUpsert   = "elasticsearch/selly.queue.order.upsert"
+	SubjectPullOrderUpsert    = "elasticsearch/selly.pull.order.upsert"
+	SubjectRequestOrderSearch = "elasticsearch/selly.request.order.search"
+
+	SubjectRequestUserUpsert = "elasticsearch/selly.request.order.upsert"
+	SubjectQueueUserUpsert   = "elasticsearch/selly.queue.order.upsert"
+	SubjectPushUserUpsert    = "elasticsearch/selly.push.order.upsert"
+	SubjectPullUserUpsert    = "elasticsearch/selly.pull.order.upsert"
+	SubjectRequestUserSearch = "elasticsearch/selly.request.user.search"
 )
 
 const (
-	JetStreamSearchService = "JetStreamSearchService"
+	JetStreamSearchService = "Service_Search"
 )
diff --git a/elasticsearch.go b/elasticsearch.go
index 1c3d6dc..c9e73d2 100644
--- a/elasticsearch.go
+++ b/elasticsearch.go
@@ -13,8 +13,16 @@ type Client struct {
 	Config        Config
 	natsServer    natsio.Server
 	natsJetStream natsio.JetStream
+	Queue         Queue
+	Pull          Pull
+	Request       Request
+	Push          Push
 }
 
+var (
+	client *Client
+)
+
 // NewClient
 // Init client elasticsearch
 func NewClient(config Config) (*Client, error) {
@@ -28,66 +36,34 @@ func NewClient(config Config) (*Client, error) {
 		return nil, fmt.Errorf("nats connect failed: %v", err)
 	}
 
-	c := &Client{
+	client = &Client{
 		Config:        config,
 		natsServer:    natsio.GetServer(),
 		natsJetStream: natsio.GetJetStream(),
+		Queue:         Queue{},
+		Pull:          Pull{},
+		Request:       Request{},
+		Push:          Push{},
 	}
 
-	return c, nil
+	return client, nil
 }
 
-// SyncData
-// Sync data to services ES
-func (c *Client) SyncData(data SyncData) (bool, error) {
+func GetClient() *Client {
+	return client
+}
+
+// RequestNats
+// publish message to nats and waiting response
+func (c *Client) RequestNats(subject string, data []byte) (*Response, error) {
 	var (
 		req = RequestBody{
 			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(data),
+			Body:   data,
 		}
 		res *Response
 	)
-	msg, err := c.natsServer.Request(SubjectSyncData, toBytes(req))
-	if err != nil {
-		return false, err
-	}
-	if err = json.Unmarshal(msg.Data, &res); err != nil {
-		return false, err
-	}
-	if res.Message != "" {
-		return false, errors.New(res.Message)
-	}
-	return res.Success, nil
-}
-
-// SyncDataWithJetStream
-// Sync data to services ES with JetStream
-func (c *Client) SyncDataWithJetStream(data SyncData) (bool, error) {
-	var (
-		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(data),
-		}
-	)
-	err := c.natsJetStream.Publish(JetStreamSearchService, SubjectSyncData, toBytes(req))
-	if err != nil {
-		return false, err
-	}
-
-	return true, nil
-}
-
-// Search
-// Request search to service es
-func (c *Client) Search(query ESQuery) (*Response, error) {
-	var (
-		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(query),
-		}
-		res *Response
-	)
-	msg, err := c.natsServer.Request(SubjectSearch, toBytes(req))
+	msg, err := c.natsServer.Request(subject, toBytes(req))
 	if err != nil {
 		return nil, err
 	}
@@ -100,61 +76,23 @@ func (c *Client) Search(query ESQuery) (*Response, error) {
 	return res, nil
 }
 
-// UpdateDocument
-// Insert or update document to ES
-func (c *Client) UpdateDocument(query UpdateDataPayload) (bool, error) {
+// PublishWithJetStream
+// Sync data to services ES with JetStream
+func (c *Client) PublishWithJetStream(streamName, subject string, data []byte) (bool, error) {
 	var (
 		req = RequestBody{
 			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(query),
+			Body:   data,
 		}
 	)
-	err := c.natsJetStream.Publish(JetStreamSearchService, SubjectUpdateDocument, toBytes(req))
+	err := c.natsJetStream.Publish(streamName, subject, toBytes(req))
 	if err != nil {
 		return false, err
 	}
-	return true, nil
-}
 
-// DeleteDocument
-// Delete document to ES
-func (c *Client) DeleteDocument(payload DeleteDataPayload) (bool, error) {
-	var (
-		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(payload),
-		}
-	)
-	err := c.natsJetStream.Publish(JetStreamSearchService, SubjectUpdateDocument, toBytes(req))
-	if err != nil {
-		return false, err
-	}
 	return true, nil
 }
 
-// CreateIndex
-// Create index to ES
-func (c *Client) CreateIndex(name string) (bool, error) {
-	var (
-		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
-			Body:   toBytes(name),
-		}
-		res *Response
-	)
-	msg, err := c.natsServer.Request(SubjectCreateIndex, toBytes(req))
-	if err != nil {
-		return false, err
-	}
-	if err = json.Unmarshal(msg.Data, &res); err != nil {
-		return false, err
-	}
-	if res.Message != "" {
-		return false, errors.New(res.Message)
-	}
-	return res.Success, nil
-}
-
 func toBytes(data interface{}) []byte {
 	b, _ := json.Marshal(data)
 	return b
diff --git a/go.mod b/go.mod
index 555267f..f612c46 100644
--- a/go.mod
+++ b/go.mod
@@ -2,13 +2,13 @@ module github.com/Selly-Modules/elasticsearch
 
 go 1.17
 
-require github.com/Selly-Modules/natsio v0.0.0-20211202032952-04a8b182fb92
+require github.com/Selly-Modules/natsio v0.0.0-20220318042849-ee34e47598b7
 
 require (
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
 	github.com/nats-io/nats.go v1.13.0 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/thoas/go-funk v0.9.1 // indirect
-	golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e // indirect
+	github.com/thoas/go-funk v0.9.2 // indirect
+	golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd // indirect
 )
diff --git a/go.sum b/go.sum
index beec991..7580e6f 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,7 @@
 github.com/Selly-Modules/natsio v0.0.0-20211202032952-04a8b182fb92 h1:tZMu1uKbo2l5G7aYRkCtyjQl4Gog4X98mYT/QZC94+8=
 github.com/Selly-Modules/natsio v0.0.0-20211202032952-04a8b182fb92/go.mod h1:NG55g9ip18nvN5tfP6PcSEKec10/lOeIOZC8HqBVNlQ=
+github.com/Selly-Modules/natsio v0.0.0-20220318042849-ee34e47598b7 h1:t4kLJt+Q4U4m79Ms9S6zjmIfTXSyrzxzVwLU6nGrQOo=
+github.com/Selly-Modules/natsio v0.0.0-20220318042849-ee34e47598b7/go.mod h1:NG55g9ip18nvN5tfP6PcSEKec10/lOeIOZC8HqBVNlQ=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
@@ -43,22 +45,29 @@ github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJy
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/thoas/go-funk v0.9.1 h1:O549iLZqPpTUQ10ykd26sZhzD+rmR5pWhuElrhbC20M=
 github.com/thoas/go-funk v0.9.1/go.mod h1:+IWnUfUmFO1+WVYQWQtIJHeRRdaIyyYglZN7xzUPe4Q=
+github.com/thoas/go-funk v0.9.2 h1:oKlNYv0AY5nyf9g+/GhMgS/UO2ces0QRdPKwkhY3VCk=
+github.com/thoas/go-funk v0.9.2/go.mod h1:+IWnUfUmFO1+WVYQWQtIJHeRRdaIyyYglZN7xzUPe4Q=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd h1:XcWmESyNjXJMLahc3mqVQJcgSTDxFxhETVlfk9uGc38=
+golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 h1:SrN+KX8Art/Sf4HNj6Zcz06G7VEz+7w9tdXTPOZ7+l4=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 h1:NusfzzA6yGQ+ua51ck7E3omNUX/JuqbFSaRGqU8CcLI=
 golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/pull.go b/pull.go
new file mode 100644
index 0000000..0d3a1b9
--- /dev/null
+++ b/pull.go
@@ -0,0 +1,20 @@
+package elasticsearch
+
+// Pull ...
+type Pull struct {
+}
+
+// ProductUpsert ...
+func (r Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
+}
+
+// UserUpsert ...
+func (r Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
+}
+
+// OrderUpsert ...
+func (r Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
+}
diff --git a/push.go b/push.go
new file mode 100644
index 0000000..7ebf746
--- /dev/null
+++ b/push.go
@@ -0,0 +1,5 @@
+package elasticsearch
+
+// Push ...
+type Push struct {
+}
diff --git a/queue.go b/queue.go
new file mode 100644
index 0000000..0c3fbbd
--- /dev/null
+++ b/queue.go
@@ -0,0 +1,20 @@
+package elasticsearch
+
+// Queue ...
+type Queue struct {
+}
+
+// ProductUpsert ...
+func (r Queue) ProductUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueProductUpsert, toBytes(query))
+}
+
+// UserUpsert ...
+func (r Queue) UserUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueUserUpsert, toBytes(query))
+}
+
+// OrderUpsert ...
+func (r Queue) OrderUpsert(query UpdateDataPayload) (bool, error) {
+	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueOrderUpsert, toBytes(query))
+}
diff --git a/request.go b/request.go
new file mode 100644
index 0000000..597969f
--- /dev/null
+++ b/request.go
@@ -0,0 +1,35 @@
+package elasticsearch
+
+// Request ...
+type Request struct {
+}
+
+// ProductSearch ...
+func (r Request) ProductSearch(query ESQuery) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestProductSearch, toBytes(query))
+}
+
+// ProductUpsert ...
+func (r Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestProductUpsert, toBytes(query))
+}
+
+// UserSearch ...
+func (r Request) UserSearch(query ESQuery) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestUserSearch, toBytes(query))
+}
+
+// UserUpsert ...
+func (r Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestUserUpsert, toBytes(query))
+}
+
+// OrderSearch ...
+func (r Request) OrderSearch(query ESQuery) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestOrderSearch, toBytes(query))
+}
+
+// OrderUpsert ...
+func (r Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
+	return GetClient().RequestNats(SubjectRequestOrderUpsert, toBytes(query))
+}

From f26bf10ea0c15f23e2c3b4d3e8e2257ba77c0531 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Sat, 19 Mar 2022 12:05:54 +0700
Subject: [PATCH 02/10] [Update] Define method

---
 constants.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/constants.go b/constants.go
index f997694..ee48263 100644
--- a/constants.go
+++ b/constants.go
@@ -13,7 +13,6 @@ const (
 
 	SubjectRequestUserUpsert = "elasticsearch/selly.request.order.upsert"
 	SubjectQueueUserUpsert   = "elasticsearch/selly.queue.order.upsert"
-	SubjectPushUserUpsert    = "elasticsearch/selly.push.order.upsert"
 	SubjectPullUserUpsert    = "elasticsearch/selly.pull.order.upsert"
 	SubjectRequestUserSearch = "elasticsearch/selly.request.user.search"
 )

From a6e7982be6f13d16a51d94ddd23fa7557ad1b052 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Sat, 19 Mar 2022 12:08:07 +0700
Subject: [PATCH 03/10] [Update] Remove queue

---
 constants.go |  3 ---
 queue.go     | 15 ---------------
 2 files changed, 18 deletions(-)

diff --git a/constants.go b/constants.go
index ee48263..7fa7ef2 100644
--- a/constants.go
+++ b/constants.go
@@ -2,17 +2,14 @@ package elasticsearch
 
 const (
 	SubjectRequestProductUpsert = "elasticsearch/selly.request.product.upsert"
-	SubjectQueueProductUpsert   = "elasticsearch/selly.queue.product.upsert"
 	SubjectPullProductUpsert    = "elasticsearch/selly.pull.product.upsert"
 	SubjectRequestProductSearch = "elasticsearch/selly.request.product.search"
 
 	SubjectRequestOrderUpsert = "elasticsearch/selly.request.order.upsert"
-	SubjectQueueOrderUpsert   = "elasticsearch/selly.queue.order.upsert"
 	SubjectPullOrderUpsert    = "elasticsearch/selly.pull.order.upsert"
 	SubjectRequestOrderSearch = "elasticsearch/selly.request.order.search"
 
 	SubjectRequestUserUpsert = "elasticsearch/selly.request.order.upsert"
-	SubjectQueueUserUpsert   = "elasticsearch/selly.queue.order.upsert"
 	SubjectPullUserUpsert    = "elasticsearch/selly.pull.order.upsert"
 	SubjectRequestUserSearch = "elasticsearch/selly.request.user.search"
 )
diff --git a/queue.go b/queue.go
index 0c3fbbd..bc1675a 100644
--- a/queue.go
+++ b/queue.go
@@ -3,18 +3,3 @@ package elasticsearch
 // Queue ...
 type Queue struct {
 }
-
-// ProductUpsert ...
-func (r Queue) ProductUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueProductUpsert, toBytes(query))
-}
-
-// UserUpsert ...
-func (r Queue) UserUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueUserUpsert, toBytes(query))
-}
-
-// OrderUpsert ...
-func (r Queue) OrderUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectQueueOrderUpsert, toBytes(query))
-}

From 37c217be50f4ab7348f06afcaca49b5cb671453d Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Sat, 19 Mar 2022 13:12:59 +0700
Subject: [PATCH 04/10] [Update] hide func

---
 elasticsearch.go | 16 ++++++++--------
 pull.go          |  6 +++---
 request.go       | 12 ++++++------
 3 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/elasticsearch.go b/elasticsearch.go
index c9e73d2..633313b 100644
--- a/elasticsearch.go
+++ b/elasticsearch.go
@@ -53,17 +53,17 @@ func GetClient() *Client {
 	return client
 }
 
-// RequestNats
+// requestNats
 // publish message to nats and waiting response
-func (c *Client) RequestNats(subject string, data []byte) (*Response, error) {
+func requestNats(subject string, data []byte) (*Response, error) {
 	var (
 		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
+			ApiKey: client.Config.ApiKey,
 			Body:   data,
 		}
 		res *Response
 	)
-	msg, err := c.natsServer.Request(subject, toBytes(req))
+	msg, err := client.natsServer.Request(subject, toBytes(req))
 	if err != nil {
 		return nil, err
 	}
@@ -76,16 +76,16 @@ func (c *Client) RequestNats(subject string, data []byte) (*Response, error) {
 	return res, nil
 }
 
-// PublishWithJetStream
+// publishWithJetStream
 // Sync data to services ES with JetStream
-func (c *Client) PublishWithJetStream(streamName, subject string, data []byte) (bool, error) {
+func publishWithJetStream(streamName, subject string, data []byte) (bool, error) {
 	var (
 		req = RequestBody{
-			ApiKey: c.Config.ApiKey,
+			ApiKey: client.Config.ApiKey,
 			Body:   data,
 		}
 	)
-	err := c.natsJetStream.Publish(streamName, subject, toBytes(req))
+	err := client.natsJetStream.Publish(streamName, subject, toBytes(req))
 	if err != nil {
 		return false, err
 	}
diff --git a/pull.go b/pull.go
index 0d3a1b9..a3d5215 100644
--- a/pull.go
+++ b/pull.go
@@ -6,15 +6,15 @@ type Pull struct {
 
 // ProductUpsert ...
 func (r Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
+	return publishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
 }
 
 // UserUpsert ...
 func (r Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
+	return publishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
 }
 
 // OrderUpsert ...
 func (r Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
-	return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
+	return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
 }
diff --git a/request.go b/request.go
index 597969f..fa59f55 100644
--- a/request.go
+++ b/request.go
@@ -6,30 +6,30 @@ type Request struct {
 
 // ProductSearch ...
 func (r Request) ProductSearch(query ESQuery) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestProductSearch, toBytes(query))
+	return requestNats(SubjectRequestProductSearch, toBytes(query))
 }
 
 // ProductUpsert ...
 func (r Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestProductUpsert, toBytes(query))
+	return requestNats(SubjectRequestProductUpsert, toBytes(query))
 }
 
 // UserSearch ...
 func (r Request) UserSearch(query ESQuery) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestUserSearch, toBytes(query))
+	return requestNats(SubjectRequestUserSearch, toBytes(query))
 }
 
 // UserUpsert ...
 func (r Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestUserUpsert, toBytes(query))
+	return requestNats(SubjectRequestUserUpsert, toBytes(query))
 }
 
 // OrderSearch ...
 func (r Request) OrderSearch(query ESQuery) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestOrderSearch, toBytes(query))
+	return requestNats(SubjectRequestOrderSearch, toBytes(query))
 }
 
 // OrderUpsert ...
 func (r Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
-	return GetClient().RequestNats(SubjectRequestOrderUpsert, toBytes(query))
+	return requestNats(SubjectRequestOrderUpsert, toBytes(query))
 }

From 5e7a5ee2c7ee78280e4ddd9388ec62890b96396a Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Sat, 19 Mar 2022 13:33:06 +0700
Subject: [PATCH 05/10] [Updaate] Method

---
 pull.go    |  6 +++---
 request.go | 12 ++++++------
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/pull.go b/pull.go
index a3d5215..4911b36 100644
--- a/pull.go
+++ b/pull.go
@@ -5,16 +5,16 @@ type Pull struct {
 }
 
 // ProductUpsert ...
-func (r Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
+func (Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
 	return publishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
 }
 
 // UserUpsert ...
-func (r Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
+func (Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
 	return publishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
 }
 
 // OrderUpsert ...
-func (r Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
+func (Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
 	return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
 }
diff --git a/request.go b/request.go
index fa59f55..d67b5c3 100644
--- a/request.go
+++ b/request.go
@@ -5,31 +5,31 @@ type Request struct {
 }
 
 // ProductSearch ...
-func (r Request) ProductSearch(query ESQuery) (*Response, error) {
+func (Request) ProductSearch(query ESQuery) (*Response, error) {
 	return requestNats(SubjectRequestProductSearch, toBytes(query))
 }
 
 // ProductUpsert ...
-func (r Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
+func (Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
 	return requestNats(SubjectRequestProductUpsert, toBytes(query))
 }
 
 // UserSearch ...
-func (r Request) UserSearch(query ESQuery) (*Response, error) {
+func (Request) UserSearch(query ESQuery) (*Response, error) {
 	return requestNats(SubjectRequestUserSearch, toBytes(query))
 }
 
 // UserUpsert ...
-func (r Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
+func (Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
 	return requestNats(SubjectRequestUserUpsert, toBytes(query))
 }
 
 // OrderSearch ...
-func (r Request) OrderSearch(query ESQuery) (*Response, error) {
+func (Request) OrderSearch(query ESQuery) (*Response, error) {
 	return requestNats(SubjectRequestOrderSearch, toBytes(query))
 }
 
 // OrderUpsert ...
-func (r Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
+func (Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
 	return requestNats(SubjectRequestOrderUpsert, toBytes(query))
 }

From 3f6d5d14f554a33e6e2c694c2c0f690613f3e030 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Mon, 21 Mar 2022 09:19:46 +0700
Subject: [PATCH 06/10] [Update] Param name

---
 constants.go | 18 +++++++++---------
 pull.go      | 12 ++++++------
 request.go   | 12 ++++++------
 struct.go    | 13 +++----------
 4 files changed, 24 insertions(+), 31 deletions(-)

diff --git a/constants.go b/constants.go
index 7fa7ef2..16ebb38 100644
--- a/constants.go
+++ b/constants.go
@@ -1,17 +1,17 @@
 package elasticsearch
 
 const (
-	SubjectRequestProductUpsert = "elasticsearch/selly.request.product.upsert"
-	SubjectPullProductUpsert    = "elasticsearch/selly.pull.product.upsert"
-	SubjectRequestProductSearch = "elasticsearch/selly.request.product.search"
+	SubjectRequestProductUpsert = "selly.request.product.upsert"
+	SubjectPullProductUpsert    = "selly.pull.product.upsert"
+	SubjectRequestProductSearch = "selly.request.product.search"
 
-	SubjectRequestOrderUpsert = "elasticsearch/selly.request.order.upsert"
-	SubjectPullOrderUpsert    = "elasticsearch/selly.pull.order.upsert"
-	SubjectRequestOrderSearch = "elasticsearch/selly.request.order.search"
+	SubjectRequestOrderUpsert = "selly.request.order.upsert"
+	SubjectPullOrderUpsert    = "selly.pull.order.upsert"
+	SubjectRequestOrderSearch = "selly.request.order.search"
 
-	SubjectRequestUserUpsert = "elasticsearch/selly.request.order.upsert"
-	SubjectPullUserUpsert    = "elasticsearch/selly.pull.order.upsert"
-	SubjectRequestUserSearch = "elasticsearch/selly.request.user.search"
+	SubjectRequestUserUpsert = "selly.request.user.upsert"
+	SubjectPullUserUpsert    = "selly.pull.user.upsert"
+	SubjectRequestUserSearch = "selly.request.user.search"
 )
 
 const (
diff --git a/pull.go b/pull.go
index 4911b36..fe354bc 100644
--- a/pull.go
+++ b/pull.go
@@ -5,16 +5,16 @@ type Pull struct {
 }
 
 // ProductUpsert ...
-func (Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
-	return publishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
+func (Pull) ProductUpsert(payload Payload) (bool, error) {
+	return publishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(payload))
 }
 
 // UserUpsert ...
-func (Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
-	return publishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
+func (Pull) UserUpsert(payload Payload) (bool, error) {
+	return publishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(payload))
 }
 
 // OrderUpsert ...
-func (Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
-	return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
+func (Pull) OrderUpsert(payload Payload) (bool, error) {
+	return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(payload))
 }
diff --git a/request.go b/request.go
index d67b5c3..3693532 100644
--- a/request.go
+++ b/request.go
@@ -10,8 +10,8 @@ func (Request) ProductSearch(query ESQuery) (*Response, error) {
 }
 
 // ProductUpsert ...
-func (Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
-	return requestNats(SubjectRequestProductUpsert, toBytes(query))
+func (Request) ProductUpsert(payload Payload) (*Response, error) {
+	return requestNats(SubjectRequestProductUpsert, toBytes(payload))
 }
 
 // UserSearch ...
@@ -20,8 +20,8 @@ func (Request) UserSearch(query ESQuery) (*Response, error) {
 }
 
 // UserUpsert ...
-func (Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
-	return requestNats(SubjectRequestUserUpsert, toBytes(query))
+func (Request) UserUpsert(payload Payload) (*Response, error) {
+	return requestNats(SubjectRequestUserUpsert, toBytes(payload))
 }
 
 // OrderSearch ...
@@ -30,6 +30,6 @@ func (Request) OrderSearch(query ESQuery) (*Response, error) {
 }
 
 // OrderUpsert ...
-func (Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
-	return requestNats(SubjectRequestOrderUpsert, toBytes(query))
+func (Request) OrderUpsert(payload Payload) (*Response, error) {
+	return requestNats(SubjectRequestOrderUpsert, toBytes(payload))
 }
diff --git a/struct.go b/struct.go
index 602f8a8..504b9b5 100644
--- a/struct.go
+++ b/struct.go
@@ -19,20 +19,13 @@ type Response struct {
 	Message string   `json:"message"`
 }
 
-// SyncData
-// Payload for sync data to service es
-type SyncData struct {
+// Payload ...
+// payload for sync data to service es
+type Payload struct {
 	Index string
 	Data  []byte
 }
 
-// UpdateDataPayload
-// Payload for insert or update document
-type UpdateDataPayload struct {
-	Index string
-	Body  []byte
-}
-
 // DeleteDataPayload
 // Payload for delete document
 type DeleteDataPayload struct {

From cf96a60a5a547abec68da2a60d3e98fc1e8aa741 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Mon, 21 Mar 2022 11:01:42 +0700
Subject: [PATCH 07/10] [Update] Clear func not use

---
 elasticsearch.go | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/elasticsearch.go b/elasticsearch.go
index 633313b..2fcf915 100644
--- a/elasticsearch.go
+++ b/elasticsearch.go
@@ -49,10 +49,6 @@ func NewClient(config Config) (*Client, error) {
 	return client, nil
 }
 
-func GetClient() *Client {
-	return client
-}
-
 // requestNats
 // publish message to nats and waiting response
 func requestNats(subject string, data []byte) (*Response, error) {

From 5b760449fc2a19eed8e0684cbb3221d90ea99a38 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Tue, 22 Mar 2022 17:09:48 +0700
Subject: [PATCH 08/10] [Update] ES Query

---
 go.sum    | 4 ----
 struct.go | 2 ++
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/go.sum b/go.sum
index 7580e6f..c180bf0 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,3 @@
-github.com/Selly-Modules/natsio v0.0.0-20211202032952-04a8b182fb92 h1:tZMu1uKbo2l5G7aYRkCtyjQl4Gog4X98mYT/QZC94+8=
-github.com/Selly-Modules/natsio v0.0.0-20211202032952-04a8b182fb92/go.mod h1:NG55g9ip18nvN5tfP6PcSEKec10/lOeIOZC8HqBVNlQ=
 github.com/Selly-Modules/natsio v0.0.0-20220318042849-ee34e47598b7 h1:t4kLJt+Q4U4m79Ms9S6zjmIfTXSyrzxzVwLU6nGrQOo=
 github.com/Selly-Modules/natsio v0.0.0-20220318042849-ee34e47598b7/go.mod h1:NG55g9ip18nvN5tfP6PcSEKec10/lOeIOZC8HqBVNlQ=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
@@ -43,14 +41,12 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
-github.com/thoas/go-funk v0.9.1 h1:O549iLZqPpTUQ10ykd26sZhzD+rmR5pWhuElrhbC20M=
 github.com/thoas/go-funk v0.9.1/go.mod h1:+IWnUfUmFO1+WVYQWQtIJHeRRdaIyyYglZN7xzUPe4Q=
 github.com/thoas/go-funk v0.9.2 h1:oKlNYv0AY5nyf9g+/GhMgS/UO2ces0QRdPKwkhY3VCk=
 github.com/thoas/go-funk v0.9.2/go.mod h1:+IWnUfUmFO1+WVYQWQtIJHeRRdaIyyYglZN7xzUPe4Q=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd h1:XcWmESyNjXJMLahc3mqVQJcgSTDxFxhETVlfk9uGc38=
 golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
diff --git a/struct.go b/struct.go
index 504b9b5..e9aa117 100644
--- a/struct.go
+++ b/struct.go
@@ -53,6 +53,8 @@ type ESQuery struct {
 	Type                  string
 	ServiceDelivery       string
 	SourceDelivery        string
+	Brands                []string
+	NoBrand               string
 	Banned                string
 	ListUser              []string
 	ListNotUser           []string

From f81c6237f75b03098bcdb936582db8bb1fa65679 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Thu, 24 Mar 2022 10:26:23 +0700
Subject: [PATCH 09/10] Add keyword subject

---
 constants.go |  4 ++++
 pull.go      |  5 +++++
 request.go   | 10 ++++++++++
 3 files changed, 19 insertions(+)

diff --git a/constants.go b/constants.go
index 16ebb38..3239e26 100644
--- a/constants.go
+++ b/constants.go
@@ -12,6 +12,10 @@ const (
 	SubjectRequestUserUpsert = "selly.request.user.upsert"
 	SubjectPullUserUpsert    = "selly.pull.user.upsert"
 	SubjectRequestUserSearch = "selly.request.user.search"
+
+	SubjectRequestKeywordUpsert = "selly.request.keyword.upsert"
+	SubjectPullKeywordUpsert    = "selly.pull.keyword.upsert"
+	SubjectRequestKeywordSearch = "selly.request.keyword.search"
 )
 
 const (
diff --git a/pull.go b/pull.go
index fe354bc..9ec9170 100644
--- a/pull.go
+++ b/pull.go
@@ -18,3 +18,8 @@ func (Pull) UserUpsert(payload Payload) (bool, error) {
 func (Pull) OrderUpsert(payload Payload) (bool, error) {
 	return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(payload))
 }
+
+// KeywordUpsert ...
+func (Pull) KeywordUpsert(payload Payload) (bool, error) {
+	return publishWithJetStream(JetStreamSearchService, SubjectPullKeywordUpsert, toBytes(payload))
+}
diff --git a/request.go b/request.go
index 3693532..743cf1a 100644
--- a/request.go
+++ b/request.go
@@ -33,3 +33,13 @@ func (Request) OrderSearch(query ESQuery) (*Response, error) {
 func (Request) OrderUpsert(payload Payload) (*Response, error) {
 	return requestNats(SubjectRequestOrderUpsert, toBytes(payload))
 }
+
+// KeywordSearch ...
+func (Request) KeywordSearch(query ESQuery) (*Response, error) {
+	return requestNats(SubjectRequestKeywordSearch, toBytes(query))
+}
+
+// KeywordUpsert ...
+func (Request) KeywordUpsert(payload Payload) (*Response, error) {
+	return requestNats(SubjectRequestKeywordUpsert, toBytes(payload))
+}

From 30bead253d7456b1957194a3fd75783647ed4d52 Mon Sep 17 00:00:00 2001
From: trunglt251292 <trunglt251292@gmail.com>
Date: Thu, 24 Mar 2022 16:48:08 +0700
Subject: [PATCH 10/10] [Update] Response

---
 struct.go | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/struct.go b/struct.go
index e9aa117..c4e5ddc 100644
--- a/struct.go
+++ b/struct.go
@@ -11,12 +11,12 @@ type RequestBody struct {
 // Response
 // response to service es
 type Response struct {
-	Success bool     `json:"success"`
-	Data    []string `json:"data,omitempty"`
-	Total   int64    `json:"total,omitempty"`
-	Page    int64    `json:"page,omitempty"`
-	Limit   int64    `json:"limit,omitempty"`
-	Message string   `json:"message"`
+	Success bool   `json:"success"`
+	Data    []byte `json:"data,omitempty"`
+	Total   int64  `json:"total,omitempty"`
+	Page    int64  `json:"page,omitempty"`
+	Limit   int64  `json:"limit,omitempty"`
+	Message string `json:"message"`
 }
 
 // Payload ...