Repository: absmach/magistrala
Pull request #112 (2025-04-10 16:04:02)

MG-659 - Add message storage to Rules Engine
<!-- Copyright (c) Abstract Machines
SPDX-License-Identifier: Apache-2.0 -->

<!--

Pull request title should be `MG-XXX - description` or `NOISSUE - description` where XXX is ID of the issue that this PR relate to.
Please review the [CONTRIBUTING.md](https://github.com/absmach/magistrala/blob/main/CONTRIBUTING.md) file for detailed contributing guidelines.

For Work In Progress Pull Requests, please use the Draft PR feature, see https://github.blog/2019-02-14-introducing-draft-pull-requests/ for further details.

For a timely review/response, please avoid force-pushing additional commits if your PR already received reviews or comments.

- Provide tests for your changes.
- Use descriptive commit messages.
- Comment your code where appropriate.
- Squash your commits
- Update any related documentation.
-->

# What type of PR is this?
This is a feature pr
<!--This represents the type of PR you are submitting.

For example:
This is a bug fix because it fixes the following issue: #1234
This is a feature because it adds the following functionality: ...
This is a refactor because it changes the following functionality: ...
This is a documentation update because it updates the following documentation: ...
This is a dependency update because it updates the following dependencies: ...
This is an optimization because it improves the following functionality: ...
-->

## What does this do?
Add message storage function to rules engine. This will allow the UI to call the function as a lua script function, allowing the storage node to publish to the writer topic
<!--
Please provide a brief description of what this PR is intended to do.
Include List any changes that modify/break current functionality.
-->

## Which issue(s) does this PR fix/relate to?
Related to https://github.com/absmach/magistrala-ui/issues/659
<!--
For pull requests that relate or close an issue, please include them below.  We like to follow [Github's guidance on linking issues to pull requests](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue).

For example having the text: "Resolves #1234" would connect the current pull request to issue 1234.  And when we merge the pull request, Github will automatically close the issue.
-->

- Related Issue #
- Resolves #

## Have you included tests for your changes?
no
<!--If you have not included tests, please explain why.
For example:
Yes, I have included tests for my changes.
No, I have not included tests because I do not know how to.
-->

## Did you document any new/modified feature?
not yet
<!--If you have not included documentation, please explain why.
For example:
Yes, I have updated the documentation for the new feature.
No, I have not updated the documentation because I do not know how to.
-->

### Notes

<!--Please provide any additional information you feel is important.-->


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/absmach/magistrala/pull/112?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) Report
Attention: Patch coverage is `28.20513%` with `28 lines` in your changes missing coverage. Please review.
> Project coverage is 27.99%. Comparing base [(`0e04e90`)](https://app.codecov.io/gh/absmach/magistrala/commit/0e04e9080341986b03ee33bb5c1f0c9453c00da0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) to head [(`31cf81b`)](https://app.codecov.io/gh/absmach/magistrala/commit/31cf81ba34ef1c59c44f4d7fa2c591c2c626dffe?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach).
> Report is 6 commits behind head on main.

| [Files with missing lines](https://app.codecov.io/gh/absmach/magistrala/pull/112?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) | Patch % | Lines |
|---|---|---|
| [re/service.go](https://app.codecov.io/gh/absmach/magistrala/pull/112?src=pr&el=tree&filepath=re%2Fservice.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvc2VydmljZS5nbw==) | 28.20% | [28 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/112?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #112      +/-   ##
==========================================
+ Coverage   27.80%   27.99%   +0.18%     
==========================================
  Files          97       17      -80     
  Lines       14880     1754   -13126     
==========================================
- Hits         4138      491    -3647     
+ Misses      10482     1239    -9243     
+ Partials      260       24     -236     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/absmach/magistrala/pull/112?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
dborovcanin: Comments will be resolved in the follow-up PR.

Patch:
diff --git a/cmd/re/main.go b/cmd/re/main.go
index 862614267..aedfa656c 100644
--- a/cmd/re/main.go
+++ b/cmd/re/main.go
@@ -31,6 +31,7 @@ import (
 	"github.com/absmach/supermq/pkg/messaging"
 	"github.com/absmach/supermq/pkg/messaging/brokers"
 	brokerstracing "github.com/absmach/supermq/pkg/messaging/brokers/tracing"
+	"github.com/absmach/supermq/pkg/messaging/nats"
 	pgclient "github.com/absmach/supermq/pkg/postgres"
 	"github.com/absmach/supermq/pkg/server"
 	httpserver "github.com/absmach/supermq/pkg/server/http"
@@ -50,6 +51,11 @@ const (
 	defSvcHTTPPort = "9008"
 )
 
+var (
+	alarmsPrefix  = "alarms"
+	writersPrefix = "writers"
+)
+
 type config struct {
 	LogLevel         string        `env:"MG_RE_LOG_LEVEL"           envDefault:"info"`
 	InstanceID       string        `env:"MG_RE_INSTANCE_ID"         envDefault:""`
@@ -124,13 +130,30 @@ func main() {
 		}
 	}()
 	tracer := tp.Tracer(svcName)
-	pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+
+	rePubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+	if err != nil {
+		logger.Error(fmt.Sprintf("failed to connect to message broker for rePubSub: %s", err))
+		exitCode = 1
+		return
+	}
+	defer rePubSub.Close()
+
+	writersPubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger, nats.Prefix(writersPrefix))
 	if err != nil {
-		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
+		logger.Error(fmt.Sprintf("failed to connect to message broker for writersPubSub: %s", err))
 		exitCode = 1
 		return
 	}
-	defer pubSub.Close()
+	defer writersPubSub.Close()
+
+	alarmsPubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger, nats.Prefix(alarmsPrefix))
+	if err != nil {
+		logger.Error(fmt.Sprintf("failed to connect to message broker for alarmsPubSub: %s", err))
+		exitCode = 1
+		return
+	}
+	defer alarmsPubSub.Close()
 
 	httpServerConfig := server.Config{Port: defSvcHTTPPort}
 	if err := env.ParseWithOptions(&httpServerConfig, env.Options{Prefix: envPrefixHTTP}); err != nil {
@@ -138,7 +161,10 @@ func main() {
 		exitCode = 1
 		return
 	}
-	pubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, pubSub)
+
+	rePubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, rePubSub)
+	writersPubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, writersPubSub)
+	alarmsPubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, alarmsPubSub)
 
 	// Setup new redis cache client
 	// cacheclient, err := redisclient.Connect(cfg.CacheURL)
@@ -174,14 +200,14 @@ func main() {
 	logger.Info("AuthZ  successfully connected to auth gRPC server " + authnClient.Secure())
 
 	database := pgclient.NewDatabase(db, dbConfig, tracer)
-	svc, err := newService(ctx, database, pubSub, authz, cfg.ESURL, tracer, ec, logger)
+	svc, err := newService(ctx, database, rePubSub, writersPubSub, alarmsPubSub, authz, cfg.ESURL, tracer, ec, logger)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to create services: %s", err))
 		exitCode = 1
 		return
 	}
 
-	if err = consumers.Start(ctx, svcName, pubSub, svc, cfg.ConfigPath, logger); err != nil {
+	if err = consumers.Start(ctx, svcName, rePubSub, svc, cfg.ConfigPath, logger); err != nil {
 		logger.Error(fmt.Sprintf("failed to create Rule Engine: %s", err))
 		exitCode = 1
 		return
@@ -214,7 +240,7 @@ func main() {
 	}
 }
 
-func newService(ctx context.Context, db pgclient.Database, pubsub messaging.PubSub, authz mgauthz.Authorization, esURL string, tracer trace.Tracer, ec email.Config, logger *slog.Logger) (re.Service, error) {
+func newService(ctx context.Context, db pgclient.Database, rePubSub messaging.PubSub, writersPubSub messaging.PubSub, alarmsPubSub messaging.PubSub, authz mgauthz.Authorization, esURL string, tracer trace.Tracer, ec email.Config, logger *slog.Logger) (re.Service, error) {
 	repo := repg.NewRepository(db)
 	idp := uuid.New()
 
@@ -224,7 +250,7 @@ func newService(ctx context.Context, db pgclient.Database, pubsub messaging.PubS
 	}
 
 	// csvc = authzmw.AuthorizationMiddleware(csvc, authz)
-	csvc := re.NewService(repo, idp, pubsub, re.NewTicker(time.Minute), emailerClient)
+	csvc := re.NewService(repo, idp, rePubSub, writersPubSub, alarmsPubSub, re.NewTicker(time.Minute), emailerClient)
 	csvc = middleware.LoggingMiddleware(csvc, logger)
 
 	return csvc, nil
diff --git a/docker/addons/postgres-writer/config.toml b/docker/addons/postgres-writer/config.toml
index b04ce56ff..40d094923 100644
--- a/docker/addons/postgres-writer/config.toml
+++ b/docker/addons/postgres-writer/config.toml
@@ -1,11 +1,11 @@
 # Copyright (c) Abstract Machines
 # SPDX-License-Identifier: Apache-2.0
 
-# To listen all messsage broker subjects use default value "channels.>".
-# To subscribe to specific subjects use values starting by "channels." and
-# followed by a subtopic (e.g ["channels.<channel_id>.sub.topic.x", ...]).
-[subscriber]
-subjects = ["channels.>"]
+# To listen all messsage broker subjects use default value "writers.>".
+# To subscribe to specific subjects use values starting by "writers." and
+# followed by a subtopic (e.g ["writers.<channel_id>.sub.topic.x", ...]).
+[subjects]
+filter = ["writers.>"]
 
 [transformer]
 # SenML or JSON
diff --git a/docker/addons/timescale-writer/config.toml b/docker/addons/timescale-writer/config.toml
index f3ad91d1e..f37228ad4 100644
--- a/docker/addons/timescale-writer/config.toml
+++ b/docker/addons/timescale-writer/config.toml
@@ -1,8 +1,8 @@
 # Copyright (c) Abstract Machines
 # SPDX-License-Identifier: Apache-2.0
 
-# To listen all messsage broker subjects use default value "channels.>".
-# To subscribe to specific subjects use values starting by "channels." and
-# followed by a subtopic (e.g ["channels.<channel_id>.sub.topic.x", ...]).
+# To listen all messsage broker subjects use default value "writers.>".
+# To subscribe to specific subjects use values starting by "writers." and
+# followed by a subtopic (e.g ["writers.<channel_id>.sub.topic.x", ...]).
 [subjects]
-filter = ["channels.>"]
+filter = ["writers.>"]
diff --git a/re/service.go b/re/service.go
index 29052e2cd..522e5e169 100644
--- a/re/service.go
+++ b/re/service.go
@@ -16,6 +16,7 @@ import (
 	"github.com/absmach/supermq/pkg/messaging"
 	"github.com/absmach/supermq/pkg/transformers"
 	mgjson "github.com/absmach/supermq/pkg/transformers/json"
+	"github.com/absmach/supermq/pkg/transformers/senml"
 	mgsenml "github.com/absmach/supermq/pkg/transformers/senml"
 	"github.com/vadv/gopher-lua-libs/argparse"
 	"github.com/vadv/gopher-lua-libs/base64"
@@ -112,23 +113,27 @@ type Service interface {
 }
 
 type re struct {
-	idp    supermq.IDProvider
-	repo   Repository
-	pubSub messaging.PubSub
-	errors chan error
-	ticker Ticker
-	email  Emailer
-	ts     []transformers.Transformer
+	writersPubSub messaging.PubSub
+	alarmsPubSub  messaging.PubSub
+	rePubSub      messaging.PubSub
+	idp           supermq.IDProvider
+	repo          Repository
+	errors        chan error
+	ticker        Ticker
+	email         Emailer
+	ts            []transformers.Transformer
 }
 
-func NewService(repo Repository, idp supermq.IDProvider, pubSub messaging.PubSub, tck Ticker, emailer Emailer) Service {
+func NewService(repo Repository, idp supermq.IDProvider, rePubSub messaging.PubSub, writersPubSub messaging.PubSub, alarmsPubSub messaging.PubSub, tck Ticker, emailer Emailer) Service {
 	return &re{
-		repo:   repo,
-		idp:    idp,
-		pubSub: pubSub,
-		errors: make(chan error),
-		ticker: tck,
-		email:  emailer,
+		writersPubSub: writersPubSub,
+		alarmsPubSub:  alarmsPubSub,
+		rePubSub:      rePubSub,
+		repo:          repo,
+		idp:           idp,
+		errors:        make(chan error),
+		ticker:        tck,
+		email:         emailer,
 		// Transformers order is important since SenML is also JSON content type.
 		ts: []transformers.Transformer{
 			mgsenml.New(mgsenml.JSON),
@@ -332,9 +337,11 @@ func (re *re) process(ctx context.Context, r Rule, msg interface{}) error {
 	// Set the message object as a Lua global variable.
 	l.SetGlobal("message", message)
 	l.SetGlobal("messages", messages)
+	l.SetGlobal("domain_id", lua.LString(r.DomainID))
 
 	// set the email function as a Lua global function
 	l.SetGlobal("send_email", l.NewFunction(re.sendEmail))
+	l.SetGlobal("save_senml", l.NewFunction(re.saveSenml))
 
 	if err := l.DoString(string(r.Logic.Value)); err != nil {
 		return err
@@ -356,7 +363,7 @@ func (re *re) process(ctx context.Context, r Rule, msg interface{}) error {
 			Domain:    r.DomainID,
 			Subtopic:  r.OutputTopic,
 		}
-		return re.pubSub.Publish(ctx, m.Channel, m)
+		return re.rePubSub.Publish(ctx, m.Channel, m)
 	}
 }
 
@@ -460,6 +467,37 @@ func (re *re) sendEmail(L *lua.LState) int {
 	return 1
 }
 
+func (re *re) saveSenml(L *lua.LState) int {
+	luaString := L.ToString(1)
+	var message senml.Message
+
+	if err := json.Unmarshal([]byte(luaString), &message); err != nil {
+		return 0
+	}
+
+	payload, err := json.Marshal(message)
+	if err != nil {
+		return 0
+	}
+
+	domainId := L.GetGlobal("domain_id").String()
+
+	ctx := context.Background()
+	m := &messaging.Message{
+		Publisher: message.Publisher,
+		Created:   time.Now().Unix(),
+		Payload:   payload,
+		Channel:   message.Channel,
+		Subtopic:  message.Subtopic,
+		Domain:    domainId,
+	}
+
+	if err := re.writersPubSub.Publish(ctx, message.Channel, m); err != nil {
+		return 0
+	}
+	return 1
+}
+
 func preload(l *lua.LState) {
 	db.Preload(l)
 	ioutil.Preload(l)


Test patch:
diff --git a/re/service_test.go b/re/service_test.go
index fc43baec5..d6f80d039 100644
--- a/re/service_test.go
+++ b/re/service_test.go
@@ -63,7 +63,7 @@ func newService(t *testing.T) (re.Service, *mocks.Repository, *pubsubmocks.PubSu
 	idProvider := uuid.NewMock()
 	pubsub := pubsubmocks.NewPubSub(t)
 	e := new(mocks.Emailer)
-	return re.NewService(repo, idProvider, pubsub, mockTicker, e), repo, pubsub, mockTicker
+	return re.NewService(repo, idProvider, pubsub, pubsub, pubsub, mockTicker, e), repo, pubsub, mockTicker
 }
 
 func TestAddRule(t *testing.T) {