Repository: absmach/magistrala
Pull request #121 (2025-04-17 20:25:26)

NOISSUE - Add script outputs
<!-- 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.
<!--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?
This adds a new field called Kind to Script struct to enable simpler result handling.
<!--
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?
There is no such issue.
<!--
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.
-->

## Have you included tests for your changes?

<!--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.
-->
No. I will do that in follow-up pull requests.

## Did you document any new/modified feature?
N/A
<!--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
N/A
<!--Please provide any additional information you feel is important.-->


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/absmach/magistrala/pull/121?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 `10.84337%` with `74 lines` in your changes missing coverage. Please review.
> Project coverage is 23.44%. Comparing base [(`513586c`)](https://app.codecov.io/gh/absmach/magistrala/commit/513586c9826a1a3ff0d8c64fccec9c7ed041b8d6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) to head [(`50ce4fc`)](https://app.codecov.io/gh/absmach/magistrala/commit/50ce4fcaf010dbd0f7c46738a255cbccbf978ef4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach).

| [Files with missing lines](https://app.codecov.io/gh/absmach/magistrala/pull/121?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/bindings.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&filepath=re%2Fbindings.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvYmluZGluZ3MuZ28=) | 0.00% | [31 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |
| [re/rule.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&filepath=re%2Frule.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvcnVsZS5nbw==) | 11.11% | [15 Missing and 1 partial :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |
| [re/postgres/init.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&filepath=re%2Fpostgres%2Finit.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvcG9zdGdyZXMvaW5pdC5nbw==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |
| [re/postgres/repository.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&filepath=re%2Fpostgres%2Frepository.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvcG9zdGdyZXMvcmVwb3NpdG9yeS5nbw==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |
| [re/service.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?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==) | 50.00% | [7 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) |
| [re/postgres/rule.go](https://app.codecov.io/gh/absmach/magistrala/pull/121?src=pr&el=tree&filepath=re%2Fpostgres%2Frule.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#diff-cmUvcG9zdGdyZXMvcnVsZS5nbw==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/absmach/magistrala/pull/121?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     #121      +/-   ##
==========================================
- Coverage   26.94%   23.44%   -3.50%     
==========================================
  Files         117       20      -97     
  Lines       16622     2073   -14549     
==========================================
- Hits         4478      486    -3992     
+ Misses      11839     1560   -10279     
+ Partials      305       27     -278     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/absmach/magistrala/pull/121?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>

Patch:
diff --git a/cmd/re/main.go b/cmd/re/main.go
index af0d3c457..f67a23b5b 100644
--- a/cmd/re/main.go
+++ b/cmd/re/main.go
@@ -176,9 +176,10 @@ func main() {
 	}
 	defer authnClient.Close()
 	logger.Info("AuthN  successfully connected to auth gRPC server " + authnClient.Secure())
+	errs := make(chan error)
 
 	database := pgclient.NewDatabase(db, dbConfig, tracer)
-	svc, err := newService(database, msgSub, writersPub, alarmsPub, ec, logger)
+	svc, err := newService(database, errs, msgSub, writersPub, alarmsPub, ec, logger)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to create services: %s", err))
 		exitCode = 1
@@ -200,8 +201,10 @@ func main() {
 
 	go func() {
 		for {
-			err := <-svc.Errors()
-			logger.Warn("Error handling rule", slog.String("error", err.Error()))
+			err := <-errs
+			if err != nil {
+				logger.Warn("Error handling rule", slog.String("error", err.Error()))
+			}
 		}
 	}()
 
@@ -231,7 +234,7 @@ func main() {
 	}
 }
 
-func newService(db pgclient.Database, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, ec email.Config, logger *slog.Logger) (re.Service, error) {
+func newService(db pgclient.Database, errs chan error, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, ec email.Config, logger *slog.Logger) (re.Service, error) {
 	repo := repg.NewRepository(db)
 	idp := uuid.New()
 
@@ -240,7 +243,7 @@ func newService(db pgclient.Database, rePubSub messaging.PubSub, writersPub, ala
 		logger.Error(fmt.Sprintf("failed to configure e-mailing util: %s", err.Error()))
 	}
 
-	csvc := re.NewService(repo, idp, rePubSub, writersPub, alarmsPub, re.NewTicker(time.Minute), emailerClient)
+	csvc := re.NewService(repo, errs, idp, rePubSub, writersPub, alarmsPub, re.NewTicker(time.Minute), emailerClient)
 	csvc = middleware.LoggingMiddleware(csvc, logger)
 
 	return csvc, nil
diff --git a/consumers/writers/timescale/init.go b/consumers/writers/timescale/init.go
index cfd7156bc..3c1f6fa5f 100644
--- a/consumers/writers/timescale/init.go
+++ b/consumers/writers/timescale/init.go
@@ -16,7 +16,7 @@ func Migration() *migrate.MemoryMigrationSource {
                         time BIGINT NOT NULL,
                         channel       UUID,
                         subtopic      VARCHAR(254),
-                        publisher     UUID,
+                        publisher     VARCHAR(254),
                         protocol      TEXT,
                         name          VARCHAR(254),
                         unit          TEXT,
@@ -26,7 +26,7 @@ func Migration() *migrate.MemoryMigrationSource {
                         data_value    BYTEA,
                         sum           FLOAT,
                         update_time   FLOAT,
-                        PRIMARY KEY (time, publisher, subtopic, name)
+                        PRIMARY KEY (time, publisher, channel, subtopic, name)
                     );
                     SELECT create_hypertable('messages', 'time', create_default_indexes => FALSE, chunk_time_interval => 86400000, if_not_exists => TRUE);`,
 				},
diff --git a/go.mod b/go.mod
index 530fb8f9e..091864f50 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.24.2
 require (
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.14.0
-	github.com/absmach/supermq v0.16.1-0.20250415080928-62f45991404c
+	github.com/absmach/supermq v0.16.1-0.20250421184358-c14bb6e18a82
 	github.com/authzed/authzed-go v1.4.0
 	github.com/authzed/grpcutil v0.0.0-20250221190651-1985b19b35b8
 	github.com/caarlos0/env/v11 v11.3.1
@@ -85,7 +85,7 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jzelinskie/stringz v0.0.3 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
-	github.com/lib/pq v1.10.9 // indirect
+	github.com/lib/pq v1.10.9
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-sqlite3 v1.14.22 // indirect
@@ -131,7 +131,7 @@ require (
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.37.0 // indirect
-	golang.org/x/net v0.37.0 // indirect
+	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
diff --git a/go.sum b/go.sum
index 8a154e798..0a54b01e3 100644
--- a/go.sum
+++ b/go.sum
@@ -28,8 +28,8 @@ github.com/absmach/mgate v0.4.5 h1:l6RmrEsR9jxkdb9WHUSecmT0HA41TkZZQVffFfUAIfI=
 github.com/absmach/mgate v0.4.5/go.mod h1:IvRIHZexZPEIAPmmaJF0L5DY2ERjj+GxRGitOW4s6qo=
 github.com/absmach/senml v1.0.7 h1:XLvpw0qxbP2QhOz7KLM2ZRar+vSCpSG/0o0kEvWx3No=
 github.com/absmach/senml v1.0.7/go.mod h1:3bRIiNc8hq7l3auMs8gQrpsM5hHy7iDuiLILrf/+MfA=
-github.com/absmach/supermq v0.16.1-0.20250415080928-62f45991404c h1:OjaoWbaRFJX42s1xyXxevuODnEyK7l2WryEfho8PWPI=
-github.com/absmach/supermq v0.16.1-0.20250415080928-62f45991404c/go.mod h1:H71WVAlZK8ljDQWkzih6iQJ2ZT/IyZUILfAxLQBzLqo=
+github.com/absmach/supermq v0.16.1-0.20250421184358-c14bb6e18a82 h1:X4zz8oLH6kGFugsuMwn8Zs9kZ1dZgDXopqPBEVSEK7I=
+github.com/absmach/supermq v0.16.1-0.20250421184358-c14bb6e18a82/go.mod h1:1EhTeUYANhKiQN4rYaoXJW+sRz2rv0AdTwPMxlPc+Hw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -577,8 +577,8 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
-golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
-golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
+golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.29.0 h1:WdYw2tdTK1S8olAzWHdgeqfy+Mtm9XNhv/xJsY65d98=
diff --git a/re/bindings.go b/re/bindings.go
index 8070cf58f..06122a73a 100644
--- a/re/bindings.go
+++ b/re/bindings.go
@@ -10,49 +10,15 @@ import (
 	"encoding/json"
 
 	"github.com/absmach/magistrala/alarms"
+	"github.com/absmach/senml"
 	"github.com/absmach/supermq/pkg/messaging"
-	"github.com/absmach/supermq/pkg/transformers/senml"
 	lua "github.com/yuin/gopher-lua"
 )
 
-func (re *re) save(ctx context.Context, original *messaging.Message) lua.LGFunction {
-	return func(l *lua.LState) int {
-		table := l.ToTable(1)
-		val := convertLua(table)
-		// In case there is a single SenML value, convert to slice so we can unmarshal.
-		if _, ok := val.([]any); !ok {
-			val = []any{val}
-		}
-		data, err := json.Marshal(val)
-		if err != nil {
-			return 0
-		}
-
-		var message []senml.Message
-		if err := json.Unmarshal(data, &message); err != nil {
-			return 0
-		}
-
-		m := &messaging.Message{
-			Domain:    original.Domain,
-			Publisher: original.Publisher,
-			Created:   original.Created,
-			Channel:   original.Channel,
-			Subtopic:  original.Subtopic,
-			Protocol:  original.Protocol,
-			Payload:   data,
-		}
-		if err := re.writersPub.Publish(ctx, original.Channel, m); err != nil {
-			return 0
-		}
-		return 1
-	}
-}
-
-func (re *re) sendEmail(L *lua.LState) int {
-	recipientsTable := L.ToTable(1)
-	subject := L.ToString(2)
-	content := L.ToString(3)
+func (re *re) sendEmail(l *lua.LState) int {
+	recipientsTable := l.ToTable(1)
+	subject := l.ToString(2)
+	content := l.ToString(3)
 
 	var recipients []string
 	recipientsTable.ForEach(func(_, value lua.LValue) {
@@ -121,3 +87,52 @@ func (re *re) sendAlarm(ctx context.Context, ruleID string, original *messaging.
 		return 1
 	}
 }
+
+func (re *re) saveSenml(ctx context.Context, val interface{}, msg *messaging.Message) error {
+	// In case there is a single SenML value, convert to slice so we can decode.
+	if _, ok := val.([]any); !ok {
+		val = []any{val}
+	}
+	data, err := json.Marshal(val)
+	if err != nil {
+		return err
+	}
+	if _, err := senml.Decode(data, senml.JSON); err != nil {
+		return err
+	}
+
+	m := &messaging.Message{
+		Domain:    msg.Domain,
+		Publisher: msg.Publisher,
+		Created:   msg.Created,
+		Channel:   msg.Channel,
+		Subtopic:  msg.Subtopic,
+		Protocol:  msg.Protocol,
+		Payload:   data,
+	}
+	if err := re.writersPub.Publish(ctx, msg.Channel, m); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (re *re) publishChannel(ctx context.Context, val interface{}, channel, subtopic string, msg *messaging.Message) error {
+	data, err := json.Marshal(val)
+	if err != nil {
+		return err
+	}
+
+	m := &messaging.Message{
+		Domain:    msg.Domain,
+		Publisher: msg.Publisher,
+		Created:   msg.Created,
+		Channel:   channel,
+		Subtopic:  subtopic,
+		Protocol:  protocol,
+		Payload:   data,
+	}
+	if err := re.rePubSub.Publish(ctx, channel, m); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/re/handlers.go b/re/handlers.go
new file mode 100644
index 000000000..3221ab8ea
--- /dev/null
+++ b/re/handlers.go
@@ -0,0 +1,178 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package re
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/absmach/supermq/pkg/errors"
+	"github.com/absmach/supermq/pkg/messaging"
+	lua "github.com/yuin/gopher-lua"
+)
+
+const (
+	maxPayload     = 100 * 1024
+	pldExceededFmt = "max payload size of 100kB exceeded: "
+)
+
+func (re *re) Handle(msg *messaging.Message) error {
+	// Limit payload for RE so we don't get to process large JSON.
+	if n := len(msg.Payload); n > maxPayload {
+		return errors.New(pldExceededFmt + strconv.Itoa(n))
+	}
+	inputChannel := msg.Channel
+	pm := PageMeta{
+		InputChannel: inputChannel,
+		Status:       EnabledStatus,
+		InputTopic:   &msg.Subtopic,
+	}
+	ctx := context.Background()
+	page, err := re.repo.ListRules(ctx, pm)
+	if err != nil {
+		return err
+	}
+
+	for _, r := range page.Rules {
+		go func(ctx context.Context) {
+			if err := re.process(ctx, r, msg); err != nil {
+				re.errors <- err
+			}
+		}(ctx)
+	}
+	return nil
+}
+
+func (re *re) process(ctx context.Context, r Rule, msg *messaging.Message) error {
+	l := lua.NewState()
+	defer l.Close()
+	preload(l)
+	message := prepareMsg(l, msg)
+
+	// Set the message object as a Lua global variable.
+	l.SetGlobal("message", message)
+
+	// set the email function as a Lua global function.
+	l.SetGlobal("send_email", l.NewFunction(re.sendEmail))
+	l.SetGlobal("send_alarm", l.NewFunction(re.sendAlarm(ctx, r.ID, msg)))
+
+	if err := l.DoString(r.Logic.Value); err != nil {
+		return err
+	}
+	// Get the last result.
+	result := l.Get(-1)
+	if result == lua.LNil {
+		return nil
+	}
+	var err error
+	for _, o := range r.Logic.Outputs {
+		val := convertLua(result)
+		// If value is false, don't run the follow-up.
+		if v, ok := val.(bool); ok && !v {
+			return nil
+		}
+		if e := re.handleOutput(ctx, o, r, msg, val); e != nil {
+			err = errors.Wrap(e, err)
+		}
+	}
+	return err
+}
+
+func (re *re) handleOutput(ctx context.Context, o ScriptOutput, r Rule, msg *messaging.Message, val interface{}) error {
+	switch o {
+	case Channels:
+		if r.OutputChannel == "" {
+			return nil
+		}
+		return re.publishChannel(ctx, val, r.OutputChannel, r.OutputTopic, msg)
+	case SaveSenML:
+		return re.saveSenml(ctx, val, msg)
+	case Email:
+		break
+	}
+	return nil
+}
+
+func (re *re) StartScheduler(ctx context.Context) error {
+	defer re.ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-re.ticker.Tick():
+			startTime := time.Now()
+
+			pm := PageMeta{
+				Status:          EnabledStatus,
+				ScheduledBefore: &startTime,
+			}
+
+			page, err := re.repo.ListRules(ctx, pm)
+			if err != nil {
+				return err
+			}
+
+			for _, rule := range page.Rules {
+				if rule.shouldRun(startTime) {
+					go func(r Rule) {
+						msg := &messaging.Message{
+							Channel: r.InputChannel,
+							Created: startTime.Unix(),
+						}
+						re.errors <- re.process(ctx, r, msg)
+					}(rule)
+				}
+			}
+		}
+	}
+}
+
+func (r Rule) shouldRun(startTime time.Time) bool {
+	// Don't run if the rule's start time is in the future
+	// This allows scheduling rules to start at a specific future time
+	if r.Schedule.StartDateTime.After(startTime) {
+		return false
+	}
+
+	t := r.Schedule.Time.Truncate(time.Minute).UTC()
+	startTimeOnly := time.Date(0, 1, 1, startTime.Hour(), startTime.Minute(), 0, 0, time.UTC)
+	if t.Equal(startTimeOnly) {
+		return true
+	}
+
+	if r.Schedule.RecurringPeriod == 0 {
+		return false
+	}
+
+	period := int(r.Schedule.RecurringPeriod)
+
+	switch r.Schedule.Recurring {
+	case Daily:
+		if r.Schedule.RecurringPeriod > 0 {
+			daysSinceStart := startTime.Sub(r.Schedule.StartDateTime).Hours() / hoursInDay
+			if int(daysSinceStart)%period == 0 {
+				return true
+			}
+		}
+	case Weekly:
+		if r.Schedule.RecurringPeriod > 0 {
+			weeksSinceStart := startTime.Sub(r.Schedule.StartDateTime).Hours() / (hoursInDay * daysInWeek)
+			if int(weeksSinceStart)%period == 0 {
+				return true
+			}
+		}
+	case Monthly:
+		if r.Schedule.RecurringPeriod > 0 {
+			monthsSinceStart := (startTime.Year()-r.Schedule.StartDateTime.Year())*monthsInYear +
+				int(startTime.Month()-r.Schedule.StartDateTime.Month())
+			if monthsSinceStart%period == 0 {
+				return true
+			}
+		}
+	}
+
+	return false
+}
diff --git a/re/middleware/logging.go b/re/middleware/logging.go
index 2b2cbefff..46415aefb 100644
--- a/re/middleware/logging.go
+++ b/re/middleware/logging.go
@@ -195,25 +195,25 @@ func (lm *loggingMiddleware) StartScheduler(ctx context.Context) (err error) {
 	return lm.svc.StartScheduler(ctx)
 }
 
-func (lm *loggingMiddleware) Handle(msg *messaging.Message) error {
+func (lm *loggingMiddleware) Handle(msg *messaging.Message) (err error) {
 	defer func(begin time.Time) {
-		args := []any{
-			slog.String("duration", time.Since(begin).String()),
-		}
-		if msg != nil {
-			args = append(args,
-				slog.String("channel", msg.Channel),
-				slog.String("payload_size", fmt.Sprintf("%d", len(msg.Payload))),
-			)
+		// Log only errors since we consume a lot of messages.
+		if err != nil {
+			args := []any{
+				slog.String("duration", time.Since(begin).String()),
+			}
+			if msg != nil {
+				args = append(args,
+					slog.String("channel", msg.Channel),
+					slog.String("payload_size", fmt.Sprintf("%d", len(msg.Payload))),
+				)
+			}
+			lm.logger.Warn("Message consumption completed", args...)
 		}
-		lm.logger.Info("Message consumption completed", args...)
 	}(time.Now())
 
-	return lm.svc.Handle(msg)
-}
-
-func (lm *loggingMiddleware) Errors() <-chan error {
-	return lm.svc.Errors()
+	err = lm.svc.Handle(msg)
+	return
 }
 
 func (lm *loggingMiddleware) Cancel() error {
diff --git a/re/mocks/service.go b/re/mocks/service.go
index 5ec46a99f..a54a53179 100644
--- a/re/mocks/service.go
+++ b/re/mocks/service.go
@@ -255,52 +255,6 @@ func (_c *Service_EnableRule_Call) RunAndReturn(run func(ctx context.Context, se
 	return _c
 }
 
-// Errors provides a mock function for the type Service
-func (_mock *Service) Errors() <-chan error {
-	ret := _mock.Called()
-
-	if len(ret) == 0 {
-		panic("no return value specified for Errors")
-	}
-
-	var r0 <-chan error
-	if returnFunc, ok := ret.Get(0).(func() <-chan error); ok {
-		r0 = returnFunc()
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(<-chan error)
-		}
-	}
-	return r0
-}
-
-// Service_Errors_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Errors'
-type Service_Errors_Call struct {
-	*mock.Call
-}
-
-// Errors is a helper method to define mock.On call
-func (_e *Service_Expecter) Errors() *Service_Errors_Call {
-	return &Service_Errors_Call{Call: _e.mock.On("Errors")}
-}
-
-func (_c *Service_Errors_Call) Run(run func()) *Service_Errors_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run()
-	})
-	return _c
-}
-
-func (_c *Service_Errors_Call) Return(errCh <-chan error) *Service_Errors_Call {
-	_c.Call.Return(errCh)
-	return _c
-}
-
-func (_c *Service_Errors_Call) RunAndReturn(run func() <-chan error) *Service_Errors_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
 // Handle provides a mock function for the type Service
 func (_mock *Service) Handle(msg *messaging.Message) error {
 	ret := _mock.Called(msg)
diff --git a/re/postgres/init.go b/re/postgres/init.go
index 902353188..843f80d19 100644
--- a/re/postgres/init.go
+++ b/re/postgres/init.go
@@ -30,7 +30,8 @@ func Migration() *migrate.MemoryMigrationSource {
 						output_channel    VARCHAR(36),
 						output_topic      TEXT,
 						status            SMALLINT NOT NULL DEFAULT 0 CHECK (status >= 0),
-						logic_type        SMALLINT NOT NULL DEFAULT 0 CHECK (status >= 0),
+						logic_type        SMALLINT NOT NULL DEFAULT 0 CHECK (logic_type >= 0),
+						logic_output 	  SMALLINT[] NOT NULL DEFAULT '{}',
 						logic_value       BYTEA,
 						time              TIMESTAMP,
 						recurring         SMALLINT,
diff --git a/re/postgres/repository.go b/re/postgres/repository.go
index da88b443c..87f043e1b 100644
--- a/re/postgres/repository.go
+++ b/re/postgres/repository.go
@@ -24,11 +24,11 @@ func NewRepository(db postgres.Database) re.Repository {
 
 func (repo *PostgresRepository) AddRule(ctx context.Context, r re.Rule) (re.Rule, error) {
 	q := `
-	INSERT INTO rules (id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value,
+	INSERT INTO rules (id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value,
 		output_channel, output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status)
-	VALUES (:id, :name, :domain_id, :metadata, :input_channel, :input_topic, :logic_type, :logic_value,
+	VALUES (:id, :name, :domain_id, :metadata, :input_channel, :input_topic, :logic_type, :logic_output, :logic_value,
 		:output_channel, :output_topic, :start_datetime, :time, :recurring, :recurring_period, :created_at, :created_by, :updated_at, :updated_by, :status)
-	RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value,
+	RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value,
 		output_channel, output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status;
 `
 	dbr, err := ruleToDb(r)
@@ -58,7 +58,7 @@ func (repo *PostgresRepository) AddRule(ctx context.Context, r re.Rule) (re.Rule
 
 func (repo *PostgresRepository) ViewRule(ctx context.Context, id string) (re.Rule, error) {
 	q := `
-		SELECT id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value, output_channel, 
+		SELECT id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value, output_channel, 
 			output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status
 		FROM rules
 		WHERE id = $1;
@@ -84,7 +84,7 @@ func (repo *PostgresRepository) UpdateRuleStatus(ctx context.Context, id string,
 		UPDATE rules
 		SET status = $2
 		WHERE id = $1
-		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value,
+		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value,
 			output_channel, output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status;  
 	`
 	row := repo.DB.QueryRowxContext(ctx, q, id, status)
@@ -127,8 +127,9 @@ func (repo *PostgresRepository) UpdateRule(ctx context.Context, r re.Rule) (re.R
 		query = append(query, "output_topic = :output_topic,")
 	}
 	if r.Logic.Value != "" {
-		query = append(query, "logic_value = :logic_value,")
 		query = append(query, "logic_type = :logic_type,")
+		query = append(query, "logic_output = :logic_output,")
+		query = append(query, "logic_value = :logic_value,")
 	}
 
 	if len(query) > 0 {
@@ -138,7 +139,7 @@ func (repo *PostgresRepository) UpdateRule(ctx context.Context, r re.Rule) (re.R
 	q := fmt.Sprintf(`
 		UPDATE rules
 		SET %s updated_at = :updated_at, updated_by = :updated_by WHERE id = :id
-		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value,
+		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value,
 			output_channel, output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status;
 	`, upq)
 
@@ -171,7 +172,7 @@ func (repo *PostgresRepository) UpdateRuleSchedule(ctx context.Context, r re.Rul
 		UPDATE rules
 		SET start_datetime = :start_datetime, time = :time, recurring = :recurring, 
 			recurring_period = :recurring_period, updated_at = :updated_at, updated_by = :updated_by WHERE id = :id
-		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_value,
+		RETURNING id, name, domain_id, metadata, input_channel, input_topic, logic_type, logic_output, logic_value,
 			output_channel, output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status;
 	`
 	dbr, err := ruleToDb(r)
@@ -230,10 +231,11 @@ func (repo *PostgresRepository) ListRules(ctx context.Context, pm re.PageMeta) (
 	}
 	pq := pageQuery(pm)
 	q := fmt.Sprintf(`
-		SELECT id, name, domain_id, input_channel, input_topic, logic_type, logic_value, output_channel, 
+		SELECT id, name, domain_id, input_channel, input_topic, logic_type, logic_output, logic_value, output_channel, 
 			output_topic, start_datetime, time, recurring, recurring_period, created_at, created_by, updated_at, updated_by, status
 		FROM rules r %s %s;
 	`, pq, pgData)
+
 	rows, err := repo.DB.NamedQueryContext(ctx, q, pm)
 	if err != nil {
 		return re.Page{}, err
@@ -274,7 +276,7 @@ func pageQuery(pm re.PageMeta) string {
 	if pm.InputChannel != "" {
 		query = append(query, "r.input_channel = :input_channel")
 	}
-	if pm.InputTopic != "" {
+	if pm.InputTopic != nil {
 		query = append(query, "r.input_topic = :input_topic")
 	}
 	if pm.OutputChannel != "" {
diff --git a/re/postgres/rule.go b/re/postgres/rule.go
index 07404b353..116c798d1 100644
--- a/re/postgres/rule.go
+++ b/re/postgres/rule.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/absmach/magistrala/re"
 	"github.com/absmach/supermq/pkg/errors"
+	"github.com/lib/pq"
 )
 
 // dbRule represents the database structure for a Rule.
@@ -21,6 +22,7 @@ type dbRule struct {
 	InputChannel    string         `db:"input_channel"`
 	InputTopic      sql.NullString `db:"input_topic"`
 	LogicType       re.ScriptType  `db:"logic_type"`
+	LogicOutputs    pq.Int32Array  `db:"logic_output"`
 	LogicValue      string         `db:"logic_value"`
 	OutputChannel   sql.NullString `db:"output_channel"`
 	OutputTopic     sql.NullString `db:"output_topic"`
@@ -44,6 +46,10 @@ func ruleToDb(r re.Rule) (dbRule, error) {
 		}
 		metadata = b
 	}
+	lo := pq.Int32Array{}
+	for _, v := range r.Logic.Outputs {
+		lo = append(lo, int32(v))
+	}
 	return dbRule{
 		ID:              r.ID,
 		Name:            r.Name,
@@ -52,6 +58,7 @@ func ruleToDb(r re.Rule) (dbRule, error) {
 		InputChannel:    r.InputChannel,
 		InputTopic:      toNullString(r.InputTopic),
 		LogicType:       r.Logic.Type,
+		LogicOutputs:    lo,
 		LogicValue:      r.Logic.Value,
 		OutputChannel:   toNullString(r.OutputChannel),
 		OutputTopic:     toNullString(r.OutputTopic),
@@ -74,6 +81,10 @@ func dbToRule(dto dbRule) (re.Rule, error) {
 			return re.Rule{}, errors.Wrap(errors.ErrMalformedEntity, err)
 		}
 	}
+	lo := []re.ScriptOutput{}
+	for _, v := range dto.LogicOutputs {
+		lo = append(lo, re.ScriptOutput(v))
+	}
 	return re.Rule{
 		ID:           dto.ID,
 		Name:         dto.Name,
@@ -82,8 +93,9 @@ func dbToRule(dto dbRule) (re.Rule, error) {
 		InputChannel: dto.InputChannel,
 		InputTopic:   fromNullString(dto.InputTopic),
 		Logic: re.Script{
-			Type:  dto.LogicType,
-			Value: dto.LogicValue,
+			Outputs: lo,
+			Type:    dto.LogicType,
+			Value:   dto.LogicValue,
 		},
 		OutputChannel: fromNullString(dto.OutputChannel),
 		OutputTopic:   fromNullString(dto.OutputTopic),
diff --git a/re/rule.go b/re/rule.go
new file mode 100644
index 000000000..ce832e2c2
--- /dev/null
+++ b/re/rule.go
@@ -0,0 +1,103 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package re
+
+import (
+	"encoding/json"
+	"strings"
+	"time"
+
+	"github.com/absmach/supermq/pkg/errors"
+)
+
+var ErrInvalidRecurringType = errors.New("invalid recurring type")
+
+const (
+	hoursInDay   = 24
+	daysInWeek   = 7
+	monthsInYear = 12
+
+	protocol = "nats"
+)
+
+// ScriptOutput is the indicator for type of the logic
+// so we can move it to the Go instead calling Go from Lua.
+type ScriptOutput uint
+
+const (
+	Channels ScriptOutput = iota
+	Alarms
+	SaveSenML
+	Email
+	SaveRemotePg
+)
+
+var (
+	scriptKindToString = [...]string{"channels", "alarms", "save_senml", "email", "save_remote_pg"}
+	stringToScriptKind = map[string]ScriptOutput{
+		"channels":       Channels,
+		"alarms":         Alarms,
+		"save_senml":     SaveSenML,
+		"email":          Email,
+		"save_remote_pg": SaveRemotePg,
+	}
+)
+
+func (s ScriptOutput) String() string {
+	if int(s) < 0 || int(s) >= len(scriptKindToString) {
+		return "unknown"
+	}
+	return scriptKindToString[s]
+}
+
+// MarshalJSON converts ScriptOutput to JSON.
+func (s ScriptOutput) MarshalJSON() ([]byte, error) {
+	return json.Marshal(s.String())
+}
+
+// UnmarshalJSON parses JSON string into ScriptOutput.
+func (s *ScriptOutput) UnmarshalJSON(data []byte) error {
+	var str string
+	if err := json.Unmarshal(data, &str); err != nil {
+		return err
+	}
+	lower := strings.ToLower(str)
+	if val, ok := stringToScriptKind[lower]; ok {
+		*s = val
+		return nil
+	}
+	return errors.New("invalid ScriptOutput: " + str)
+}
+
+type (
+	// ScriptType indicates Runtime type for the future versions
+	// that will support JS or Go runtimes alongside Lua.
+	ScriptType uint
+
+	Metadata map[string]interface{}
+
+	Script struct {
+		Type    ScriptType     `json:"type"`
+		Outputs []ScriptOutput `json:"outputs"`
+		Value   string         `json:"value"`
+	}
+)
+
+type Rule struct {
+	ID            string    `json:"id"`
+	Name          string    `json:"name"`
+	DomainID      string    `json:"domain"`
+	Metadata      Metadata  `json:"metadata,omitempty"`
+	InputChannel  string    `json:"input_channel"`
+	InputTopic    string    `json:"input_topic"`
+	Logic         Script    `json:"logic"`
+	OutputChannel string    `json:"output_channel,omitempty"`
+	OutputTopic   string    `json:"output_topic,omitempty"`
+	Schedule      Schedule  `json:"schedule,omitempty"`
+	Status        Status    `json:"status"`
+	CreatedAt     time.Time `json:"created_at,omitempty"`
+	CreatedBy     string    `json:"created_by,omitempty"`
+	UpdatedAt     time.Time `json:"updated_at,omitempty"`
+	UpdatedBy     string    `json:"updated_by,omitempty"`
+}
diff --git a/re/service.go b/re/service.go
index de1b738e8..970a6fabd 100644
--- a/re/service.go
+++ b/re/service.go
@@ -12,46 +12,8 @@ import (
 	"github.com/absmach/supermq/pkg/errors"
 	svcerr "github.com/absmach/supermq/pkg/errors/service"
 	"github.com/absmach/supermq/pkg/messaging"
-	lua "github.com/yuin/gopher-lua"
 )
 
-const (
-	hoursInDay   = 24
-	daysInWeek   = 7
-	monthsInYear = 12
-
-	publisher = "magistrala.re"
-)
-
-var ErrInvalidRecurringType = errors.New("invalid recurring type")
-
-type (
-	ScriptType uint
-	Metadata   map[string]interface{}
-	Script     struct {
-		Type  ScriptType `json:"type"`
-		Value string     `json:"value"`
-	}
-)
-
-type Rule struct {
-	ID            string    `json:"id"`
-	Name          string    `json:"name"`
-	DomainID      string    `json:"domain"`
-	Metadata      Metadata  `json:"metadata,omitempty"`
-	InputChannel  string    `json:"input_channel"`
-	InputTopic    string    `json:"input_topic"`
-	Logic         Script    `json:"logic"`
-	OutputChannel string    `json:"output_channel,omitempty"`
-	OutputTopic   string    `json:"output_topic,omitempty"`
-	Schedule      Schedule  `json:"schedule,omitempty"`
-	Status        Status    `json:"status"`
-	CreatedAt     time.Time `json:"created_at,omitempty"`
-	CreatedBy     string    `json:"created_by,omitempty"`
-	UpdatedAt     time.Time `json:"updated_at,omitempty"`
-	UpdatedBy     string    `json:"updated_by,omitempty"`
-}
-
 type Repository interface {
 	AddRule(ctx context.Context, r Rule) (Rule, error)
 	ViewRule(ctx context.Context, id string) (Rule, error)
@@ -70,7 +32,7 @@ type PageMeta struct {
 	Dir             string     `json:"dir" db:"dir"`
 	Name            string     `json:"name" db:"name"`
 	InputChannel    string     `json:"input_channel,omitempty" db:"input_channel"`
-	InputTopic      string     `json:"input_topic,omitempty" db:"input_topic"`
+	InputTopic      *string    `json:"input_topic,omitempty" db:"input_topic"`
 	OutputChannel   string     `json:"output_channel,omitempty" db:"output_channel"`
 	Status          Status     `json:"status,omitempty" db:"status"`
 	Domain          string     `json:"domain_id,omitempty" db:"domain_id"`
@@ -97,28 +59,27 @@ type Service interface {
 	EnableRule(ctx context.Context, session authn.Session, id string) (Rule, error)
 	DisableRule(ctx context.Context, session authn.Session, id string) (Rule, error)
 	StartScheduler(ctx context.Context) error
-	Errors() <-chan error
 }
 
 type re struct {
-	writersPub messaging.Publisher
-	alarmsPub  messaging.Publisher
-	rePubSub   messaging.PubSub
-	idp        supermq.IDProvider
 	repo       Repository
 	errors     chan error
+	idp        supermq.IDProvider
+	rePubSub   messaging.PubSub
+	writersPub messaging.Publisher
+	alarmsPub  messaging.Publisher
 	ticker     Ticker
 	email      Emailer
 }
 
-func NewService(repo Repository, idp supermq.IDProvider, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, tck Ticker, emailer Emailer) Service {
+func NewService(repo Repository, errors chan (error), idp supermq.IDProvider, rePubSub messaging.PubSub, writersPub, alarmsPub messaging.Publisher, tck Ticker, emailer Emailer) Service {
 	return &re{
-		writersPub: writersPub,
-		alarmsPub:  alarmsPub,
-		rePubSub:   rePubSub,
 		repo:       repo,
 		idp:        idp,
-		errors:     make(chan error),
+		errors:     errors,
+		rePubSub:   rePubSub,
+		writersPub: writersPub,
+		alarmsPub:  alarmsPub,
 		ticker:     tck,
 		email:      emailer,
 	}
@@ -220,30 +181,6 @@ func (re *re) DisableRule(ctx context.Context, session authn.Session, id string)
 	return rule, nil
 }
 
-func (re *re) Handle(msg *messaging.Message) error {
-	inputChannel := msg.Channel
-
-	pm := PageMeta{
-		InputChannel: inputChannel,
-		Status:       EnabledStatus,
-		InputTopic:   msg.Subtopic,
-	}
-	ctx := context.Background()
-	page, err := re.repo.ListRules(ctx, pm)
-	if err != nil {
-		return err
-	}
-
-	for _, r := range page.Rules {
-		go func(ctx context.Context) {
-			if err := re.process(ctx, r, msg); err != nil {
-				re.errors <- err
-			}
-		}(ctx)
-	}
-	return nil
-}
-
 func (re *re) Cancel() error {
 	return nil
 }
@@ -251,124 +188,3 @@ func (re *re) Cancel() error {
 func (re *re) Errors() <-chan error {
 	return re.errors
 }
-
-func (re *re) process(ctx context.Context, r Rule, msg *messaging.Message) error {
-	l := lua.NewState()
-	defer l.Close()
-	preload(l)
-
-	message := prepareMsg(l, msg)
-
-	// Set the message object as a Lua global variable.
-	l.SetGlobal("message", message)
-
-	// set the email function as a Lua global function.
-	l.SetGlobal("send_email", l.NewFunction(re.sendEmail))
-	l.SetGlobal("save_senml", l.NewFunction(re.save(ctx, msg)))
-	l.SetGlobal("send_alarm", l.NewFunction(re.sendAlarm(ctx, r.ID, msg)))
-
-	if err := l.DoString(string(r.Logic.Value)); err != nil {
-		return err
-	}
-
-	result := l.Get(-1) // Get the last result.
-	switch result {
-	case lua.LNil:
-		return nil
-	default:
-		if r.OutputChannel == "" {
-			return nil
-		}
-		m := &messaging.Message{
-			Publisher: publisher,
-			Created:   time.Now().Unix(),
-			Payload:   []byte(result.String()),
-			Channel:   r.OutputChannel,
-			Domain:    r.DomainID,
-			Subtopic:  r.OutputTopic,
-		}
-		return re.rePubSub.Publish(ctx, m.Channel, m)
-	}
-}
-
-func (re *re) StartScheduler(ctx context.Context) error {
-	defer re.ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-re.ticker.Tick():
-			startTime := time.Now()
-
-			pm := PageMeta{
-				Status:          EnabledStatus,
-				ScheduledBefore: &startTime,
-			}
-
-			page, err := re.repo.ListRules(ctx, pm)
-			if err != nil {
-				return err
-			}
-
-			for _, rule := range page.Rules {
-				if rule.shouldRun(startTime) {
-					go func(r Rule) {
-						msg := &messaging.Message{
-							Channel: r.InputChannel,
-							Created: startTime.Unix(),
-						}
-						re.errors <- re.process(ctx, r, msg)
-					}(rule)
-				}
-			}
-		}
-	}
-}
-
-func (r Rule) shouldRun(startTime time.Time) bool {
-	// Don't run if the rule's start time is in the future
-	// This allows scheduling rules to start at a specific future time
-	if r.Schedule.StartDateTime.After(startTime) {
-		return false
-	}
-
-	t := r.Schedule.Time.Truncate(time.Minute).UTC()
-	startTimeOnly := time.Date(0, 1, 1, startTime.Hour(), startTime.Minute(), 0, 0, time.UTC)
-	if t.Equal(startTimeOnly) {
-		return true
-	}
-
-	if r.Schedule.RecurringPeriod == 0 {
-		return false
-	}
-
-	period := int(r.Schedule.RecurringPeriod)
-
-	switch r.Schedule.Recurring {
-	case Daily:
-		if r.Schedule.RecurringPeriod > 0 {
-			daysSinceStart := startTime.Sub(r.Schedule.StartDateTime).Hours() / hoursInDay
-			if int(daysSinceStart)%period == 0 {
-				return true
-			}
-		}
-	case Weekly:
-		if r.Schedule.RecurringPeriod > 0 {
-			weeksSinceStart := startTime.Sub(r.Schedule.StartDateTime).Hours() / (hoursInDay * daysInWeek)
-			if int(weeksSinceStart)%period == 0 {
-				return true
-			}
-		}
-	case Monthly:
-		if r.Schedule.RecurringPeriod > 0 {
-			monthsSinceStart := (startTime.Year()-r.Schedule.StartDateTime.Year())*monthsInYear +
-				int(startTime.Month()-r.Schedule.StartDateTime.Month())
-			if monthsSinceStart%period == 0 {
-				return true
-			}
-		}
-	}
-
-	return false
-}


Test patch:
diff --git a/re/service_test.go b/re/service_test.go
index a8f693e93..f72a09865 100644
--- a/re/service_test.go
+++ b/re/service_test.go
@@ -56,17 +56,17 @@ var (
 	}
 )
 
-func newService(t *testing.T) (re.Service, *mocks.Repository, *pubsubmocks.PubSub, *mocks.Ticker) {
+func newService(t *testing.T, errs chan error) (re.Service, *mocks.Repository, *pubsubmocks.PubSub, *mocks.Ticker) {
 	repo := new(mocks.Repository)
 	mockTicker := new(mocks.Ticker)
 	idProvider := uuid.NewMock()
 	pubsub := pubsubmocks.NewPubSub(t)
 	e := new(mocks.Emailer)
-	return re.NewService(repo, idProvider, pubsub, pubsub, pubsub, mockTicker, e), repo, pubsub, mockTicker
+	return re.NewService(repo, errs, idProvider, pubsub, pubsub, pubsub, mockTicker, e), repo, pubsub, mockTicker
 }
 
 func TestAddRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 	ruleName := namegen.Generate()
 	now := time.Now().Add(time.Hour)
 	cases := []struct {
@@ -141,7 +141,7 @@ func TestAddRule(t *testing.T) {
 }
 
 func TestViewRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 
 	now := time.Now().Add(time.Hour)
 	cases := []struct {
@@ -199,7 +199,7 @@ func TestViewRule(t *testing.T) {
 }
 
 func TestUpdateRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 
 	newName := namegen.Generate()
 	now := time.Now().Add(time.Hour)
@@ -284,7 +284,7 @@ func TestUpdateRule(t *testing.T) {
 }
 
 func TestListRules(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 	numRules := 50
 	now := time.Now().Add(time.Hour)
 	var rules []re.Rule
@@ -389,7 +389,7 @@ func TestListRules(t *testing.T) {
 }
 
 func TestRemoveRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 
 	cases := []struct {
 		desc    string
@@ -429,7 +429,7 @@ func TestRemoveRule(t *testing.T) {
 }
 
 func TestEnableRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 
 	cases := []struct {
 		desc    string
@@ -484,7 +484,7 @@ func TestEnableRule(t *testing.T) {
 }
 
 func TestDisableRule(t *testing.T) {
-	svc, repo, _, _ := newService(t)
+	svc, repo, _, _ := newService(t, make(chan error))
 
 	cases := []struct {
 		desc    string
@@ -539,8 +539,9 @@ func TestDisableRule(t *testing.T) {
 }
 
 func TestHandle(t *testing.T) {
-	svc, repo, pubmocks, _ := newService(t)
+	svc, repo, pubmocks, _ := newService(t, make(chan error))
 	now := time.Now()
+	empty := ""
 
 	cases := []struct {
 		desc       string
@@ -559,6 +560,7 @@ func TestHandle(t *testing.T) {
 			},
 			pageMeta: re.PageMeta{
 				InputChannel: inputChannel,
+				InputTopic:   &empty,
 				Status:       re.EnabledStatus,
 			},
 			page: re.Page{
@@ -575,6 +577,7 @@ func TestHandle(t *testing.T) {
 			pageMeta: re.PageMeta{
 				InputChannel: inputChannel,
 				Status:       re.EnabledStatus,
+				InputTopic:   &empty,
 			},
 			page: re.Page{
 				Rules: []re.Rule{
@@ -619,7 +622,8 @@ func TestHandle(t *testing.T) {
 
 func TestStartScheduler(t *testing.T) {
 	now := time.Now().Truncate(time.Minute)
-	svc, repo, _, ticker := newService(t)
+	errs := make(chan error)
+	svc, repo, _, ticker := newService(t, errs)
 
 	noRecurringPeriod := re.Rule{
 		ID:           testsutil.GenerateUUID(t),
@@ -834,7 +838,7 @@ func TestStartScheduler(t *testing.T) {
 			case "start scheduler with list error":
 				tickChan <- time.Now()
 				time.Sleep(100 * time.Millisecond)
-				if err := svc.Errors(); err != nil {
+				if err := errs; err != nil {
 					cancel()
 				}
 			default: