Repository: absmach/magistrala
Pull request #84 (2025-03-10 11:06:50)

Bump go.opentelemetry.io/otel/trace from 1.34.0 to 1.35.0
Bumps [go.opentelemetry.io/otel/trace](https://github.com/open-telemetry/opentelemetry-go) from 1.34.0 to 1.35.0.
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/open-telemetry/opentelemetry-go/blob/main/CHANGELOG.md">go.opentelemetry.io/otel/trace's changelog</a>.</em></p>
<blockquote>
<h2>[1.35.0/0.57.0/0.11.0] 2025-03-05</h2>
<p>This release is the last to support [Go 1.22].
The next release will require at least [Go 1.23].</p>
<h3>Added</h3>
<ul>
<li>Add <code>ValueFromAttribute</code> and <code>KeyValueFromAttribute</code> in <code>go.opentelemetry.io/otel/log</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6180">#6180</a>)</li>
<li>Add <code>EventName</code> and <code>SetEventName</code> to <code>Record</code> in <code>go.opentelemetry.io/otel/log</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6187">#6187</a>)</li>
<li>Add <code>EventName</code> to <code>RecordFactory</code> in <code>go.opentelemetry.io/otel/log/logtest</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6187">#6187</a>)</li>
<li><code>AssertRecordEqual</code> in <code>go.opentelemetry.io/otel/log/logtest</code> checks <code>Record.EventName</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6187">#6187</a>)</li>
<li>Add <code>EventName</code> and <code>SetEventName</code> to <code>Record</code> in <code>go.opentelemetry.io/otel/sdk/log</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6193">#6193</a>)</li>
<li>Add <code>EventName</code> to <code>RecordFactory</code> in <code>go.opentelemetry.io/otel/sdk/log/logtest</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6193">#6193</a>)</li>
<li>Emit <code>Record.EventName</code> field in <code>go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6211">#6211</a>)</li>
<li>Emit <code>Record.EventName</code> field in <code>go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6211">#6211</a>)</li>
<li>Emit <code>Record.EventName</code> field in <code>go.opentelemetry.io/otel/exporters/stdout/stdoutlog</code> (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6210">#6210</a>)</li>
<li>The <code>go.opentelemetry.io/otel/semconv/v1.28.0</code> package.
The package contains semantic conventions from the <code>v1.28.0</code> version of the OpenTelemetry Semantic Conventions.
See the <a href="https://github.com/open-telemetry/opentelemetry-go/blob/main/semconv/v1.28.0/MIGRATION.md">migration documentation</a> for information on how to upgrade from <code>go.opentelemetry.io/otel/semconv/v1.27.0</code>(<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6236">#6236</a>)</li>
<li>The <code>go.opentelemetry.io/otel/semconv/v1.30.0</code> package.
The package contains semantic conventions from the <code>v1.30.0</code> version of the OpenTelemetry Semantic Conventions.
See the <a href="https://github.com/open-telemetry/opentelemetry-go/blob/main/semconv/v1.30.0/MIGRATION.md">migration documentation</a> for information on how to upgrade from <code>go.opentelemetry.io/otel/semconv/v1.28.0</code>(<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6240">#6240</a>)</li>
<li>Document the pitfalls of using <code>Resource</code> as a comparable type.
<code>Resource.Equal</code> and <code>Resource.Equivalent</code> should be used instead. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6272">#6272</a>)</li>
<li>Support [Go 1.24]. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6304">#6304</a>)</li>
<li>Add <code>FilterProcessor</code> and <code>EnabledParameters</code> in <code>go.opentelemetry.io/otel/sdk/log</code>.
It replaces <code>go.opentelemetry.io/otel/sdk/log/internal/x.FilterProcessor</code>.
Compared to previous version it additionally gives the possibility to filter by resource and instrumentation scope. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6317">#6317</a>)</li>
</ul>
<h3>Changed</h3>
<ul>
<li>Update <code>github.com/prometheus/common</code> to <code>v0.62.0</code>, which changes the <code>NameValidationScheme</code> to <code>NoEscaping</code>.
This allows metrics names to keep original delimiters (e.g. <code>.</code>), rather than replacing with underscores.
This is controlled by the <code>Content-Type</code> header, or can be reverted by setting <code>NameValidationScheme</code> to <code>LegacyValidation</code> in <code>github.com/prometheus/common/model</code>. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6198">#6198</a>)</li>
</ul>
<h3>Fixes</h3>
<ul>
<li>Eliminate goroutine leak for the processor returned by <code>NewSimpleSpanProcessor</code> in <code>go.opentelemetry.io/otel/sdk/trace</code> when <code>Shutdown</code> is called and the passed <code>ctx</code> is canceled and <code>SpanExporter.Shutdown</code> has not returned. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6368">#6368</a>)</li>
<li>Eliminate goroutine leak for the processor returned by <code>NewBatchSpanProcessor</code> in <code>go.opentelemetry.io/otel/sdk/trace</code> when <code>ForceFlush</code> is called and the passed <code>ctx</code> is canceled and <code>SpanExporter.Export</code> has not returned. (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6369">#6369</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/5ba5e7a449f36c1c02710bbaa517263797046db0"><code>5ba5e7a</code></a> Release v1.35.0/v0.57.0/v0.11.0 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6407">#6407</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/3908b67340a7d9ee5c475cc5683535bc5da51ed8"><code>3908b67</code></a> chore(deps): update module github.com/securego/gosec/v2 to v2.22.2 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6412">#6412</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/50172b1934dac5afd1527939a771b1e954624bea"><code>50172b1</code></a> chore(deps): update module github.com/ryancurrah/gomodguard to v1.4.1 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6411">#6411</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/cea6d2b039590d9a489ab836e0609cc9d7645148"><code>cea6d2b</code></a> fix(deps): update module google.golang.org/grpc to v1.71.0 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6409">#6409</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/e2aee3aff8abf61e7a8794b4d55e179ef1ed717c"><code>e2aee3a</code></a> Move trace sdk tests from trace_test into trace package (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6400">#6400</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/38f4f39284a2e7ff3b09b9afa47657a41e4af591"><code>38f4f39</code></a> fix(deps): update build-tools to v0.20.0 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6403">#6403</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/2911449bab33757c10bb32096bd76b946315a988"><code>2911449</code></a> Look at stale issues in ascending order (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6396">#6396</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/7cb322a315d6be2c9f17ddac190bac26c3cf5465"><code>7cb322a</code></a> chore(deps): update github.com/golangci/dupl digest to 44c6a0b (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6398">#6398</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/0c3651e263b1b073718ed3db49a701ee8933721f"><code>0c3651e</code></a> fix(deps): update module github.com/golangci/golangci-lint to v1.64.6 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6394">#6394</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-go/commit/f04e9512fff3e3d1dfe0c3df81ef8d0693f18696"><code>f04e951</code></a> chore(deps): update mvdan.cc/unparam digest to 0df0534 (<a href="https://redirect.github.com/open-telemetry/opentelemetry-go/issues/6391">#6391</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/open-telemetry/opentelemetry-go/compare/v1.34.0...v1.35.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=go.opentelemetry.io/otel/trace&package-manager=go_modules&previous-version=1.34.0&new-version=1.35.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/absmach/magistrala/pull/84?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach) Report
All modified and coverable lines are covered by tests :white_check_mark:
> :exclamation: No coverage uploaded for pull request base (`main@94c5a36`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=absmach#section-missing-base-commit).


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


```diff
@@           Coverage Diff           @@
##             main      #84   +/-   ##
=======================================
  Coverage        ?   34.76%           
=======================================
  Files           ?       17           
  Lines           ?      630           
  Branches        ?        0           
=======================================
  Hits            ?      219           
  Misses          ?      402           
  Partials        ?        9           
```



</details>

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


Patch:
diff --git a/cmd/re/main.go b/cmd/re/main.go
index 298069257..741f649f5 100644
--- a/cmd/re/main.go
+++ b/cmd/re/main.go
@@ -33,6 +33,7 @@ import (
 	httpserver "github.com/absmach/supermq/pkg/server/http"
 	"github.com/absmach/supermq/pkg/uuid"
 	"github.com/caarlos0/env/v11"
+	"github.com/go-chi/chi/v5"
 	"github.com/jmoiron/sqlx"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
@@ -175,7 +176,9 @@ func main() {
 		exitCode = 1
 		return
 	}
-	httpSvc := httpserver.NewServer(ctx, cancel, svcName, httpServerConfig, httpapi.MakeHandler(svc, authn, logger, cfg.InstanceID), logger)
+	mux := chi.NewRouter()
+
+	httpSvc := httpserver.NewServer(ctx, cancel, svcName, httpServerConfig, httpapi.MakeHandler(svc, authn, mux, logger, cfg.InstanceID), logger)
 
 	if cfg.SendTelemetry {
 		chc := chclient.New(svcName, supermq.Version, logger, cancel)
diff --git a/go.mod b/go.mod
index 688b64506..b1e71d19f 100644
--- a/go.mod
+++ b/go.mod
@@ -31,8 +31,8 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/yuin/gopher-lua v1.1.1
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0
-	go.opentelemetry.io/otel v1.34.0
-	go.opentelemetry.io/otel/trace v1.34.0
+	go.opentelemetry.io/otel v1.35.0
+	go.opentelemetry.io/otel/trace v1.35.0
 	golang.org/x/sync v0.11.0
 	gonum.org/v1/gonum v0.15.1
 	google.golang.org/grpc v1.70.0
@@ -124,7 +124,7 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.34.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.34.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
diff --git a/go.sum b/go.sum
index ce94f218d..d8fb0c8b0 100644
--- a/go.sum
+++ b/go.sum
@@ -136,8 +136,8 @@ github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaW
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
-github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
@@ -417,20 +417,20 @@ go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.5
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0/go.mod h1:ijPqXp5P6IRRByFVVg9DY8P5HkxkHE5ARIa+86aXPf4=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 h1:CV7UdSGJt/Ao6Gp4CXckLxVRRsRgDHoI8XjbL3PDl8s=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0/go.mod h1:FRmFuRJfag1IZ2dPkHnEoSFVgTVPUd2qf5Vi69hLb8I=
-go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
-go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
+go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ=
+go.opentelemetry.io/otel v1.35.0/go.mod h1:UEqy8Zp11hpkUrL73gSlELM0DupHoiq72dR+Zqel/+Y=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.34.0 h1:OeNbIYk/2C15ckl7glBlOBp5+WlYsOElzTNmiPW/x60=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.34.0/go.mod h1:7Bept48yIeqxP2OZ9/AqIpYS94h2or0aB4FypJTc8ZM=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.34.0 h1:BEj3SPM81McUZHYjRS5pEgNgnmzGJ5tRpU5krWnV8Bs=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.34.0/go.mod h1:9cKLGBDzI/F3NoHLQGm4ZrYdIHsvGt6ej6hUowxY0J4=
-go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ=
-go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE=
+go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/siN90M=
+go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE=
 go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
 go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
 go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiyYCU9snn1CU=
 go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ=
-go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
-go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
+go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs=
+go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc=
 go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4=
 go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
diff --git a/re/api/transport.go b/re/api/transport.go
index 75ea51748..357025d23 100644
--- a/re/api/transport.go
+++ b/re/api/transport.go
@@ -30,11 +30,10 @@ const (
 )
 
 // MakeHandler creates an HTTP handler for the service endpoints.
-func MakeHandler(svc re.Service, authn mgauthn.Authentication, logger *slog.Logger, instanceID string) http.Handler {
+func MakeHandler(svc re.Service, authn mgauthn.Authentication, mux *chi.Mux, logger *slog.Logger, instanceID string) http.Handler {
 	opts := []kithttp.ServerOption{
 		kithttp.ServerErrorEncoder(apiutil.LoggingErrorEncoder(logger, api.EncodeError)),
 	}
-	mux := chi.NewRouter()
 	mux.Group(func(r chi.Router) {
 		r.Use(api.AuthenticateMiddleware(authn, true))
 		r.Route("/{domainID}/rules", func(r chi.Router) {
@@ -112,6 +111,9 @@ func decodeViewRuleRequest(_ context.Context, r *http.Request) (interface{}, err
 }
 
 func decodeUpdateRuleRequest(_ context.Context, r *http.Request) (interface{}, error) {
+	if !strings.Contains(r.Header.Get("Content-Type"), api.ContentType) {
+		return nil, errors.Wrap(apiutil.ErrValidation, apiutil.ErrUnsupportedContentType)
+	}
 	var rule re.Rule
 	if err := json.NewDecoder(r.Body).Decode(&rule); err != nil {
 		return nil, err
@@ -148,6 +150,10 @@ func decodeListRulesRequest(_ context.Context, r *http.Request) (interface{}, er
 	if err != nil {
 		return nil, errors.Wrap(apiutil.ErrValidation, err)
 	}
+	dir, err := apiutil.ReadStringQuery(r, api.DirKey, api.DefDir)
+	if err != nil {
+		return nil, errors.Wrap(apiutil.ErrValidation, err)
+	}
 	st, err := re.ToStatus(s)
 	if err != nil {
 		return nil, errors.Wrap(apiutil.ErrValidation, err)
@@ -159,6 +165,7 @@ func decodeListRulesRequest(_ context.Context, r *http.Request) (interface{}, er
 			InputChannel:  ic,
 			OutputChannel: oc,
 			Status:        st,
+			Dir:           dir,
 		},
 	}, nil
 }
diff --git a/re/service.go b/re/service.go
index 539cc7313..46e9c4c24 100644
--- a/re/service.go
+++ b/re/service.go
@@ -5,12 +5,13 @@ package re
 
 import (
 	"context"
-	"errors"
 	"time"
 
 	"github.com/absmach/supermq"
 	"github.com/absmach/supermq/consumers"
 	"github.com/absmach/supermq/pkg/authn"
+	"github.com/absmach/supermq/pkg/errors"
+	svcerr "github.com/absmach/supermq/pkg/errors/service"
 	"github.com/absmach/supermq/pkg/messaging"
 	mgjson "github.com/absmach/supermq/pkg/transformers/json"
 	lua "github.com/yuin/gopher-lua"
@@ -131,29 +132,47 @@ func (re *re) AddRule(ctx context.Context, session authn.Session, r Rule) (Rule,
 
 	rule, err := re.repo.AddRule(ctx, r)
 	if err != nil {
-		return Rule{}, err
+		return Rule{}, errors.Wrap(svcerr.ErrCreateEntity, err)
 	}
 
 	return rule, nil
 }
 
 func (re *re) ViewRule(ctx context.Context, session authn.Session, id string) (Rule, error) {
-	return re.repo.ViewRule(ctx, id)
+	rule, err := re.repo.ViewRule(ctx, id)
+	if err != nil {
+		return Rule{}, errors.Wrap(svcerr.ErrViewEntity, err)
+	}
+
+	return rule, nil
 }
 
 func (re *re) UpdateRule(ctx context.Context, session authn.Session, r Rule) (Rule, error) {
 	r.UpdatedAt = time.Now()
 	r.UpdatedBy = session.UserID
-	return re.repo.UpdateRule(ctx, r)
+	rule, err := re.repo.UpdateRule(ctx, r)
+	if err != nil {
+		return Rule{}, errors.Wrap(svcerr.ErrUpdateEntity, err)
+	}
+
+	return rule, nil
 }
 
 func (re *re) ListRules(ctx context.Context, session authn.Session, pm PageMeta) (Page, error) {
 	pm.Domain = session.DomainID
-	return re.repo.ListRules(ctx, pm)
+	page, err := re.repo.ListRules(ctx, pm)
+	if err != nil {
+		return Page{}, errors.Wrap(svcerr.ErrViewEntity, err)
+	}
+	return page, nil
 }
 
 func (re *re) RemoveRule(ctx context.Context, session authn.Session, id string) error {
-	return re.repo.RemoveRule(ctx, id)
+	if err := re.repo.RemoveRule(ctx, id); err != nil {
+		return errors.Wrap(svcerr.ErrRemoveEntity, err)
+	}
+
+	return nil
 }
 
 func (re *re) EnableRule(ctx context.Context, session authn.Session, id string) (Rule, error) {
@@ -161,7 +180,11 @@ func (re *re) EnableRule(ctx context.Context, session authn.Session, id string)
 	if err != nil {
 		return Rule{}, err
 	}
-	return re.repo.UpdateRuleStatus(ctx, id, status)
+	rule, err := re.repo.UpdateRuleStatus(ctx, id, status)
+	if err != nil {
+		return Rule{}, errors.Wrap(svcerr.ErrUpdateEntity, err)
+	}
+	return rule, nil
 }
 
 func (re *re) DisableRule(ctx context.Context, session authn.Session, id string) (Rule, error) {
@@ -169,7 +192,11 @@ func (re *re) DisableRule(ctx context.Context, session authn.Session, id string)
 	if err != nil {
 		return Rule{}, err
 	}
-	return re.repo.UpdateRuleStatus(ctx, id, status)
+	rule, err := re.repo.UpdateRuleStatus(ctx, id, status)
+	if err != nil {
+		return Rule{}, errors.Wrap(svcerr.ErrUpdateEntity, err)
+	}
+	return rule, nil
 }
 
 func (re *re) ConsumeAsync(ctx context.Context, msgs interface{}) {
@@ -181,7 +208,7 @@ func (re *re) ConsumeAsync(ctx context.Context, msgs interface{}) {
 		}
 		page, err := re.repo.ListRules(ctx, pm)
 		if err != nil {
-			re.errors <- err
+			re.errors <- errors.Wrap(svcerr.ErrViewEntity, err)
 			return
 		}
 		for _, r := range page.Rules {
@@ -282,7 +309,7 @@ func (r Rule) shouldRun(startTime time.Time) bool {
 		return false
 	}
 
-	t := r.Schedule.Time.Truncate(time.Minute)
+	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


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index a3d20e308..dbb740ab9 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -110,6 +110,11 @@ jobs:
               - "things/**"
               - "auth/**"
 
+            re:
+              - "re/**"
+              - "cmd/re/**"
+              - "re/api/**"
+
       - name: Create coverage directory
         run: |
           mkdir coverage
@@ -154,6 +159,12 @@ jobs:
         run: |
           go test --race -v -count=1 -coverprofile=coverage/readers.out ./readers/...
 
+      - name: Run rule engine tests
+        if: steps.changes.outputs.re == 'true' || steps.changes.outputs.workflow == 'true'
+        run: |
+            go test --race -v -count=1 -coverprofile=coverage/re.out ./re/...
+  
+      
       - name: Upload coverage
         uses: codecov/codecov-action@v5
         with:
diff --git a/re/api/endpoints_test.go b/re/api/endpoints_test.go
new file mode 100644
index 000000000..1b310b8f8
--- /dev/null
+++ b/re/api/endpoints_test.go
@@ -0,0 +1,997 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package api_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/0x6flab/namegenerator"
+	"github.com/absmach/magistrala/internal/testsutil"
+	"github.com/absmach/magistrala/pkg/errors"
+	"github.com/absmach/magistrala/re"
+	"github.com/absmach/magistrala/re/api"
+	"github.com/absmach/magistrala/re/mocks"
+	apiutil "github.com/absmach/supermq/api/http/util"
+	"github.com/absmach/supermq/auth"
+	smqlog "github.com/absmach/supermq/logger"
+	smqauthn "github.com/absmach/supermq/pkg/authn"
+	authnmocks "github.com/absmach/supermq/pkg/authn/mocks"
+	svcerr "github.com/absmach/supermq/pkg/errors/service"
+	"github.com/go-chi/chi/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+const contentType = "application/json"
+
+var (
+	namegen      = namegenerator.NewGenerator()
+	domainID     = testsutil.GenerateUUID(&testing.T{})
+	userID       = testsutil.GenerateUUID(&testing.T{})
+	validID      = testsutil.GenerateUUID(&testing.T{})
+	validToken   = "valid"
+	invalidToken = "invalid"
+	now          = time.Now().UTC().Truncate(time.Minute)
+	schedule     = re.Schedule{
+		StartDateTime:   now.Add(-1 * time.Hour),
+		Recurring:       re.Daily,
+		RecurringPeriod: 1,
+		Time:            now,
+	}
+	rule = re.Rule{
+		ID:       validID,
+		Name:     namegen.Generate(),
+		DomainID: domainID,
+		Schedule: schedule,
+		Metadata: re.Metadata{
+			"name": "test",
+		},
+	}
+)
+
+type testRequest struct {
+	client      *http.Client
+	method      string
+	url         string
+	contentType string
+	token       string
+	body        io.Reader
+}
+
+func (tr testRequest) make() (*http.Response, error) {
+	req, err := http.NewRequest(tr.method, tr.url, tr.body)
+	if err != nil {
+		return nil, err
+	}
+
+	if tr.token != "" {
+		req.Header.Set("Authorization", apiutil.BearerPrefix+tr.token)
+	}
+
+	if tr.contentType != "" {
+		req.Header.Set("Content-Type", tr.contentType)
+	}
+
+	req.Header.Set("Referer", "http://localhost")
+
+	return tr.client.Do(req)
+}
+
+func newRuleEngineServer() (*httptest.Server, *mocks.Service, *authnmocks.Authentication) {
+	svc := new(mocks.Service)
+	authn := new(authnmocks.Authentication)
+
+	logger := smqlog.NewMock()
+	mux := chi.NewRouter()
+	api.MakeHandler(svc, authn, mux, logger, "")
+
+	return httptest.NewServer(mux), svc, authn
+}
+
+func toJSON(data any) string {
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		return ""
+	}
+	return string(jsonData)
+}
+
+func TestAddRuleEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	cases := []struct {
+		desc        string
+		rule        re.Rule
+		domainID    string
+		token       string
+		contentType string
+		status      int
+		authnRes    smqauthn.Session
+		authnErr    error
+		svcRes      re.Rule
+		svcErr      error
+		err         error
+		len         int
+	}{
+		{
+			desc:        "add rule successfully",
+			rule:        rule,
+			token:       validToken,
+			contentType: contentType,
+			domainID:    domainID,
+			authnRes:    smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID},
+			status:      http.StatusCreated,
+			svcRes:      rule,
+		},
+		{
+			desc:        "add rule with invalid token",
+			rule:        rule,
+			token:       invalidToken,
+			authnRes:    smqauthn.Session{},
+			domainID:    domainID,
+			contentType: contentType,
+			authnErr:    svcerr.ErrAuthentication,
+			status:      http.StatusUnauthorized,
+			err:         svcerr.ErrAuthentication,
+		},
+		{
+			desc:        "add rule with empty token",
+			token:       "",
+			authnRes:    smqauthn.Session{},
+			domainID:    domainID,
+			rule:        rule,
+			contentType: contentType,
+			status:      http.StatusUnauthorized,
+			err:         apiutil.ErrBearerToken,
+		},
+		{
+			desc:  "add rule with name that is too long",
+			token: validToken,
+			rule: re.Rule{
+				ID:   validID,
+				Name: strings.Repeat("a", 1025),
+				Logic: re.Script{
+					Type:  re.ScriptType(0),
+					Value: "return `test` end",
+				},
+			},
+			domainID:    domainID,
+			contentType: contentType,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrNameSize,
+		},
+		{
+			desc:        "add rule with empty domainID",
+			token:       validToken,
+			rule:        rule,
+			contentType: contentType,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:        "add rule with invalid content type",
+			token:       validToken,
+			domainID:    domainID,
+			rule:        rule,
+			contentType: "application/xml",
+			status:      http.StatusUnsupportedMediaType,
+			err:         apiutil.ErrUnsupportedContentType,
+		},
+		{
+			desc:        "add rule with service error",
+			token:       validToken,
+			domainID:    domainID,
+			authnRes:    smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID},
+			rule:        rule,
+			contentType: contentType,
+			svcErr:      svcerr.ErrAuthorization,
+			status:      http.StatusForbidden,
+			err:         svcerr.ErrAuthorization,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			data := toJSON(tc.rule)
+			req := testRequest{
+				client:      ts.Client(),
+				method:      http.MethodPost,
+				url:         fmt.Sprintf("%s/%s/rules", ts.URL, tc.domainID),
+				contentType: tc.contentType,
+				token:       tc.token,
+				body:        strings.NewReader(data),
+			}
+
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.authnRes, tc.authnErr)
+			svcCall := svc.On("AddRule", mock.Anything, tc.authnRes, tc.rule).Return(tc.svcRes, tc.svcErr)
+			res, err := req.make()
+
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var errRes respBody
+			err = json.NewDecoder(res.Body).Decode(&errRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if errRes.Err != "" || errRes.Message != "" {
+				err = errors.Wrap(errors.New(errRes.Err), errors.New(errRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestViewRuleEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	cases := []struct {
+		desc        string
+		id          string
+		domainID    string
+		token       string
+		contentType string
+		status      int
+		authnRes    smqauthn.Session
+		authnErr    error
+		svcRes      re.Rule
+		svcErr      error
+		err         error
+		len         int
+	}{
+		{
+			desc:        "view rule successfully",
+			id:          rule.ID,
+			token:       validToken,
+			contentType: contentType,
+			domainID:    domainID,
+			authnRes:    smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID},
+			status:      http.StatusOK,
+			svcRes:      rule,
+		},
+		{
+			desc:        "view rule with invalid token",
+			id:          rule.ID,
+			token:       invalidToken,
+			authnRes:    smqauthn.Session{},
+			domainID:    domainID,
+			contentType: contentType,
+			authnErr:    svcerr.ErrAuthentication,
+			status:      http.StatusUnauthorized,
+			err:         svcerr.ErrAuthentication,
+		},
+		{
+			desc:        "view rule with empty token",
+			token:       "",
+			authnRes:    smqauthn.Session{},
+			domainID:    domainID,
+			id:          rule.ID,
+			contentType: contentType,
+			status:      http.StatusUnauthorized,
+			err:         apiutil.ErrBearerToken,
+		},
+		{
+			desc:        "view rule with empty domainID",
+			token:       validToken,
+			id:          rule.ID,
+			contentType: contentType,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:        "view rule with service error",
+			token:       validToken,
+			domainID:    domainID,
+			authnRes:    smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID},
+			id:          rule.ID,
+			contentType: contentType,
+			svcErr:      svcerr.ErrAuthorization,
+			status:      http.StatusForbidden,
+			err:         svcerr.ErrAuthorization,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := testRequest{
+				client: ts.Client(),
+				method: http.MethodGet,
+				url:    fmt.Sprintf("%s/%s/rules/%s", ts.URL, tc.domainID, tc.id),
+				token:  tc.token,
+			}
+
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.authnRes, tc.authnErr)
+			svcCall := svc.On("ViewRule", mock.Anything, tc.authnRes, tc.id).Return(tc.svcRes, tc.svcErr)
+			res, err := req.make()
+
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var errRes respBody
+			err = json.NewDecoder(res.Body).Decode(&errRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if errRes.Err != "" || errRes.Message != "" {
+				err = errors.Wrap(errors.New(errRes.Err), errors.New(errRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestListRulesEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	cases := []struct {
+		desc              string
+		query             string
+		domainID          string
+		token             string
+		session           smqauthn.Session
+		listRulesResponse re.Page
+		status            int
+		authnErr          error
+		err               error
+	}{
+		{
+			desc:     "list rules successfully",
+			domainID: domainID,
+			token:    validToken,
+			status:   http.StatusOK,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			err: nil,
+		},
+		{
+			desc:     "list rules with empty token",
+			domainID: domainID,
+			token:    "",
+			status:   http.StatusUnauthorized,
+			err:      apiutil.ErrBearerToken,
+		},
+		{
+			desc:     "list rules with invalid token",
+			domainID: domainID,
+			token:    invalidToken,
+			status:   http.StatusUnauthorized,
+			authnErr: svcerr.ErrAuthentication,
+			err:      svcerr.ErrAuthentication,
+		},
+		{
+			desc:     "list rules with offset",
+			domainID: domainID,
+			token:    validToken,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			query:  "offset=1",
+			status: http.StatusOK,
+			err:    nil,
+		},
+		{
+			desc:     "list rules with invalid offset",
+			domainID: domainID,
+			token:    validToken,
+			query:    "offset=invalid",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrValidation,
+		},
+		{
+			desc:     "list rules with limit",
+			domainID: domainID,
+			token:    validToken,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			query:  "limit=1",
+			status: http.StatusOK,
+			err:    nil,
+		},
+		{
+			desc:     "list rules with invalid limit",
+			domainID: domainID,
+			token:    validToken,
+			query:    "limit=invalid",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrValidation,
+		},
+		{
+			desc:     "list rules with invalid direction",
+			domainID: domainID,
+			token:    validToken,
+			query:    "dir=invalid",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrInvalidDirection,
+		},
+		{
+			desc:     "list rule with limit that is too big",
+			domainID: domainID,
+			token:    validToken,
+			query:    "limit=10000",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrLimitSize,
+		},
+		{
+			desc:     "list rules with input channel",
+			domainID: domainID,
+			token:    validToken,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			query:  "input_channel=input.channel",
+			status: http.StatusOK,
+			err:    nil,
+		},
+		{
+			desc:     "list rules with duplicate input_channel",
+			domainID: domainID,
+			token:    validToken,
+			query:    "input_channel=1&input_channel=2",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrInvalidQueryParams,
+		},
+		{
+			desc:     "list rules with status",
+			domainID: domainID,
+			token:    validToken,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			query:  "status=enabled",
+			status: http.StatusOK,
+			err:    nil,
+		},
+		{
+			desc:     "list rules with invalid status",
+			domainID: domainID,
+			token:    validToken,
+			query:    "status=invalid",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrValidation,
+		},
+		{
+			desc:     "list rules with duplicate status",
+			domainID: domainID,
+			token:    validToken,
+			query:    "status=enabled&status=disabled",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrInvalidQueryParams,
+		},
+		{
+			desc:     "list rules with output channel",
+			domainID: domainID,
+			token:    validToken,
+			listRulesResponse: re.Page{
+				PageMeta: re.PageMeta{
+					Total: 1,
+				},
+				Rules: []re.Rule{rule},
+			},
+			query:  "output_channel=output.channel",
+			status: http.StatusOK,
+			err:    nil,
+		},
+		{
+			desc:     "list rules with duplicate output channel",
+			domainID: domainID,
+			token:    validToken,
+			query:    "output_channel=1&output_channel=2",
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrInvalidQueryParams,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := testRequest{
+				client:      ts.Client(),
+				method:      http.MethodGet,
+				url:         ts.URL + "/" + tc.domainID + "/rules?" + tc.query,
+				contentType: contentType,
+				token:       tc.token,
+			}
+			if tc.token == validToken {
+				tc.session = smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID}
+			}
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authnErr)
+			svcCall := svc.On("ListRules", mock.Anything, tc.session, mock.Anything).Return(tc.listRulesResponse, tc.err)
+			res, err := req.make()
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var bodyRes respBody
+			err = json.NewDecoder(res.Body).Decode(&bodyRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if bodyRes.Err != "" || bodyRes.Message != "" {
+				err = errors.Wrap(errors.New(bodyRes.Err), errors.New(bodyRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestUpdateRulesEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	updateRuleReq := re.Rule{
+		ID:   rule.ID,
+		Name: rule.Name,
+		Logic: re.Script{
+			Type:  re.ScriptType(0),
+			Value: "return `test` end",
+		},
+		Metadata: map[string]any{
+			"name": "test",
+		},
+	}
+
+	invalidReq := re.Rule{
+		ID:   rule.ID,
+		Name: rule.Name,
+		Metadata: map[string]any{
+			"name": "test",
+		},
+	}
+
+	cases := []struct {
+		desc        string
+		token       string
+		id          string
+		domainID    string
+		updateReq   re.Rule
+		contentType string
+		session     smqauthn.Session
+		svcResp     re.Rule
+		svcErr      error
+		status      int
+		authnErr    error
+		err         error
+	}{
+		{
+			desc:        "update rule successfully",
+			token:       validToken,
+			domainID:    domainID,
+			id:          rule.ID,
+			updateReq:   updateRuleReq,
+			contentType: contentType,
+			svcResp:     rule,
+			status:      http.StatusOK,
+			err:         nil,
+		},
+		{
+			desc:        "update rule with invalid token",
+			token:       invalidToken,
+			session:     smqauthn.Session{},
+			domainID:    domainID,
+			id:          rule.ID,
+			updateReq:   updateRuleReq,
+			contentType: contentType,
+			authnErr:    svcerr.ErrAuthentication,
+			status:      http.StatusUnauthorized,
+			err:         svcerr.ErrAuthentication,
+		},
+		{
+			desc:        "update rule with empty token",
+			token:       "",
+			session:     smqauthn.Session{},
+			domainID:    domainID,
+			id:          rule.ID,
+			updateReq:   updateRuleReq,
+			contentType: contentType,
+			status:      http.StatusUnauthorized,
+			err:         apiutil.ErrBearerToken,
+		},
+		{
+			desc:        "update rule with empty domainID",
+			token:       validToken,
+			id:          rule.ID,
+			updateReq:   updateRuleReq,
+			contentType: contentType,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:        "update rule with empty logic",
+			token:       validToken,
+			domainID:    domainID,
+			id:          rule.ID,
+			updateReq:   invalidReq,
+			contentType: contentType,
+			svcResp:     rule,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrEmptyList,
+		},
+		{
+			desc:     "update rule with name that is too long",
+			token:    validToken,
+			id:       validID,
+			domainID: domainID,
+			updateReq: re.Rule{
+				ID:   validID,
+				Name: strings.Repeat("a", 1025),
+				Logic: re.Script{
+					Type:  re.ScriptType(0),
+					Value: "return `test` end",
+				},
+			},
+			contentType: contentType,
+			status:      http.StatusBadRequest,
+			err:         apiutil.ErrNameSize,
+		},
+		{
+			desc:        "update rule with invalid content type",
+			token:       validToken,
+			id:          rule.ID,
+			domainID:    domainID,
+			updateReq:   updateRuleReq,
+			contentType: "application/xml",
+			svcResp:     rule,
+			status:      http.StatusUnsupportedMediaType,
+			err:         apiutil.ErrUnsupportedContentType,
+		},
+		{
+			desc:        "update rule with service error",
+			token:       validToken,
+			id:          rule.ID,
+			domainID:    domainID,
+			updateReq:   updateRuleReq,
+			contentType: contentType,
+			svcResp:     re.Rule{},
+			svcErr:      svcerr.ErrAuthorization,
+			status:      http.StatusForbidden,
+			err:         svcerr.ErrAuthorization,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			data := toJSON(tc.updateReq)
+			req := testRequest{
+				client:      ts.Client(),
+				method:      http.MethodPatch,
+				url:         fmt.Sprintf("%s/%s/rules/%s", ts.URL, tc.domainID, tc.id),
+				contentType: tc.contentType,
+				token:       tc.token,
+				body:        strings.NewReader(data),
+			}
+			if tc.token == validToken {
+				tc.session = smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID}
+			}
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authnErr)
+			svcCall := svc.On("UpdateRule", mock.Anything, tc.session, tc.updateReq).Return(tc.svcResp, tc.svcErr)
+			res, err := req.make()
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var errRes respBody
+			err = json.NewDecoder(res.Body).Decode(&errRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if errRes.Err != "" || errRes.Message != "" {
+				err = errors.Wrap(errors.New(errRes.Err), errors.New(errRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestEnableRuleEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	cases := []struct {
+		desc     string
+		token    string
+		id       string
+		domainID string
+		session  smqauthn.Session
+		svcResp  re.Rule
+		svcErr   error
+		status   int
+		authnErr error
+		err      error
+	}{
+		{
+			desc:     "enable rule successfully",
+			token:    validToken,
+			domainID: domainID,
+			id:       validID,
+			svcResp:  rule,
+			svcErr:   nil,
+			status:   http.StatusOK,
+			err:      nil,
+		},
+		{
+			desc:     "enable rule with invalid token",
+			token:    invalidToken,
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			authnErr: svcerr.ErrAuthentication,
+			status:   http.StatusUnauthorized,
+			err:      svcerr.ErrAuthentication,
+		},
+		{
+			desc:     "enable rule with empty token",
+			token:    "",
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			status:   http.StatusUnauthorized,
+			err:      apiutil.ErrBearerToken,
+		},
+		{
+			desc:   "enable rule with empty domainID",
+			token:  validToken,
+			id:     validID,
+			status: http.StatusBadRequest,
+			err:    apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:     "enable rule with service error",
+			token:    validToken,
+			id:       validID,
+			domainID: domainID,
+			svcResp:  re.Rule{},
+			svcErr:   svcerr.ErrAuthorization,
+			status:   http.StatusForbidden,
+			err:      svcerr.ErrAuthorization,
+		},
+		{
+			desc:     "enable rule with empty id",
+			token:    validToken,
+			id:       "",
+			domainID: domainID,
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrMissingID,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := testRequest{
+				client: ts.Client(),
+				method: http.MethodPost,
+				url:    fmt.Sprintf("%s/%s/rules/%s/enable", ts.URL, tc.domainID, tc.id),
+				token:  tc.token,
+			}
+			if tc.token == validToken {
+				tc.session = smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID}
+			}
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authnErr)
+			svcCall := svc.On("EnableRule", mock.Anything, tc.session, tc.id).Return(tc.svcResp, tc.svcErr)
+			res, err := req.make()
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var errRes respBody
+			err = json.NewDecoder(res.Body).Decode(&errRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if errRes.Err != "" || errRes.Message != "" {
+				err = errors.Wrap(errors.New(errRes.Err), errors.New(errRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestDisableRuleEndpoint(t *testing.T) {
+	gs, svc, authn := newRuleEngineServer()
+	defer gs.Close()
+
+	cases := []struct {
+		desc     string
+		token    string
+		id       string
+		domainID string
+		session  smqauthn.Session
+		svcResp  re.Rule
+		svcErr   error
+		status   int
+		authnErr error
+		err      error
+	}{
+		{
+			desc:     "disable rule successfully",
+			token:    validToken,
+			domainID: domainID,
+			id:       validID,
+			svcResp:  rule,
+			svcErr:   nil,
+			status:   http.StatusOK,
+			err:      nil,
+		},
+		{
+			desc:     "disable rule with invalid token",
+			token:    invalidToken,
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			authnErr: svcerr.ErrAuthentication,
+			status:   http.StatusUnauthorized,
+			err:      svcerr.ErrAuthentication,
+		},
+		{
+			desc:     "disable rule with empty token",
+			token:    "",
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			status:   http.StatusUnauthorized,
+			err:      apiutil.ErrBearerToken,
+		},
+		{
+			desc:   "disable rule with empty domainID",
+			token:  validToken,
+			id:     validID,
+			status: http.StatusBadRequest,
+			err:    apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:     "disable rule with service error",
+			token:    validToken,
+			id:       validID,
+			domainID: domainID,
+			svcResp:  re.Rule{},
+			svcErr:   svcerr.ErrAuthorization,
+			status:   http.StatusForbidden,
+			err:      svcerr.ErrAuthorization,
+		},
+		{
+			desc:     "disable rule with empty id",
+			token:    validToken,
+			id:       "",
+			domainID: domainID,
+			status:   http.StatusBadRequest,
+			err:      apiutil.ErrMissingID,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := testRequest{
+				client: gs.Client(),
+				method: http.MethodPost,
+				url:    fmt.Sprintf("%s/%s/rules/%s/disable", gs.URL, tc.domainID, tc.id),
+				token:  tc.token,
+			}
+			if tc.token == validToken {
+				tc.session = smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID}
+			}
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authnErr)
+			svcCall := svc.On("DisableRule", mock.Anything, tc.session, tc.id).Return(tc.svcResp, tc.svcErr)
+			res, err := req.make()
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			var errRes respBody
+			err = json.NewDecoder(res.Body).Decode(&errRes)
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error while decoding response body: %s", tc.desc, err))
+			if errRes.Err != "" || errRes.Message != "" {
+				err = errors.Wrap(errors.New(errRes.Err), errors.New(errRes.Message))
+			}
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+func TestDeleteRuleEndpoint(t *testing.T) {
+	ts, svc, authn := newRuleEngineServer()
+	defer ts.Close()
+
+	cases := []struct {
+		desc     string
+		token    string
+		id       string
+		domainID string
+		session  smqauthn.Session
+		svcErr   error
+		status   int
+		authnErr error
+		err      error
+	}{
+		{
+			desc:     "delete rule successfully",
+			token:    validToken,
+			domainID: domainID,
+			id:       validID,
+			svcErr:   nil,
+			status:   http.StatusNoContent,
+			err:      nil,
+		},
+		{
+			desc:     "delete rule with invalid token",
+			token:    invalidToken,
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			authnErr: svcerr.ErrAuthentication,
+			status:   http.StatusUnauthorized,
+			err:      svcerr.ErrAuthentication,
+		},
+		{
+			desc:     "delete rule with empty token",
+			token:    "",
+			session:  smqauthn.Session{},
+			domainID: domainID,
+			id:       validID,
+			status:   http.StatusUnauthorized,
+			err:      apiutil.ErrBearerToken,
+		},
+		{
+			desc:   "delete rule with empty domainID",
+			token:  validToken,
+			id:     validID,
+			status: http.StatusBadRequest,
+			err:    apiutil.ErrMissingDomainID,
+		},
+		{
+			desc:     "delete rule with service error",
+			token:    validToken,
+			id:       validID,
+			domainID: domainID,
+			svcErr:   svcerr.ErrAuthorization,
+			status:   http.StatusForbidden,
+			err:      svcerr.ErrAuthorization,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := testRequest{
+				client: ts.Client(),
+				method: http.MethodDelete,
+				url:    fmt.Sprintf("%s/%s/rules/%s", ts.URL, tc.domainID, tc.id),
+				token:  tc.token,
+			}
+			if tc.token == validToken {
+				tc.session = smqauthn.Session{DomainUserID: auth.EncodeDomainUserID(domainID, userID), UserID: userID, DomainID: domainID}
+			}
+			authCall := authn.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authnErr)
+			svcCall := svc.On("RemoveRule", mock.Anything, tc.session, tc.id).Return(tc.svcErr)
+			res, err := req.make()
+			assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", tc.desc, err))
+			assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
+			svcCall.Unset()
+			authCall.Unset()
+		})
+	}
+}
+
+type respBody struct {
+	Err     string    `json:"error"`
+	Message string    `json:"message"`
+	Total   uint64    `json:"total"`
+	ID      string    `json:"id"`
+	Status  re.Status `json:"status"`
+}
diff --git a/re/service_test.go b/re/service_test.go
index 97756b21b..ad08d45cb 100644
--- a/re/service_test.go
+++ b/re/service_test.go
@@ -11,34 +11,44 @@ import (
 
 	"github.com/0x6flab/namegenerator"
 	"github.com/absmach/magistrala/internal/testsutil"
-	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/absmach/magistrala/re"
 	"github.com/absmach/magistrala/re/mocks"
+	"github.com/absmach/supermq/pkg/authn"
+	"github.com/absmach/supermq/pkg/errors"
 	repoerr "github.com/absmach/supermq/pkg/errors/repository"
+	svcerr "github.com/absmach/supermq/pkg/errors/service"
+	"github.com/absmach/supermq/pkg/messaging"
 	pubsubmocks "github.com/absmach/supermq/pkg/messaging/mocks"
+	mgjson "github.com/absmach/supermq/pkg/transformers/json"
 	"github.com/absmach/supermq/pkg/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
 
 var (
-	namegen = namegenerator.NewGenerator()
-	rule    = re.Rule{
+	namegen      = namegenerator.NewGenerator()
+	userID       = testsutil.GenerateUUID(&testing.T{})
+	domainID     = testsutil.GenerateUUID(&testing.T{})
+	ruleName     = namegen.Generate()
+	ruleID       = testsutil.GenerateUUID(&testing.T{})
+	inputChannel = "test.channel"
+	schedule     = re.Schedule{
+		StartDateTime:   time.Now().Add(-time.Hour),
+		Recurring:       re.Daily,
+		RecurringPeriod: 1,
+		Time:            time.Now().Add(-time.Hour),
+	}
+	rule = re.Rule{
 		ID:           testsutil.GenerateUUID(&testing.T{}),
 		Name:         namegen.Generate(),
-		InputChannel: "test.channel",
+		InputChannel: inputChannel,
 		Status:       re.EnabledStatus,
-		Schedule: re.Schedule{
-			StartDateTime:   time.Now().Add(-time.Hour), // Started an hour ago
-			Recurring:       re.Daily,
-			RecurringPeriod: 1,
-			Time:            time.Now().Add(-time.Hour),
-		},
+		Schedule:     schedule,
 	}
 	futureRule = re.Rule{
 		ID:           testsutil.GenerateUUID(&testing.T{}),
 		Name:         namegen.Generate(),
-		InputChannel: "test.channel",
+		InputChannel: inputChannel,
 		Status:       re.EnabledStatus,
 		Schedule: re.Schedule{
 			StartDateTime: time.Now().Add(24 * time.Hour),
@@ -47,17 +57,648 @@ var (
 	}
 )
 
-func newService(t *testing.T) (re.Service, *mocks.Repository, *mocks.Ticker) {
+func newService(t *testing.T) (re.Service, *mocks.Repository, *pubsubmocks.PubSub, *mocks.Ticker) {
 	repo := new(mocks.Repository)
 	mockTicker := new(mocks.Ticker)
 	idProvider := uuid.NewMock()
 	pubsub := pubsubmocks.NewPubSub(t)
-	return re.NewService(repo, idProvider, pubsub, mockTicker), repo, mockTicker
+	return re.NewService(repo, idProvider, pubsub, mockTicker), repo, pubsub, mockTicker
+}
+
+func TestAddRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+	ruleName := namegen.Generate()
+	now := time.Now().Add(time.Hour)
+	cases := []struct {
+		desc    string
+		session authn.Session
+		rule    re.Rule
+		res     re.Rule
+		err     error
+	}{
+		{
+			desc: "Add rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			rule: re.Rule{
+				Name:         ruleName,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+			},
+			res: re.Rule{
+				Name:         ruleName,
+				ID:           ruleID,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+				Status:    re.EnabledStatus,
+				CreatedBy: userID,
+				DomainID:  domainID,
+			},
+			err: nil,
+		},
+		{
+			desc: "Add rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			rule: re.Rule{
+				Name:         ruleName,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+			},
+			err: repoerr.ErrCreateEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("AddRule", mock.Anything, mock.Anything).Return(tc.res, tc.err)
+			res, err := svc.AddRule(context.Background(), tc.session, tc.rule)
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.NotEmpty(t, res.ID, "expected non-empty result in ID")
+				assert.Equal(t, tc.rule.Name, res.Name)
+				assert.Equal(t, tc.rule.Schedule, res.Schedule)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestViewRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+
+	now := time.Now().Add(time.Hour)
+	cases := []struct {
+		desc    string
+		session authn.Session
+		id      string
+		res     re.Rule
+		err     error
+	}{
+		{
+			desc: "view rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id: ruleID,
+			res: re.Rule{
+				Name:         ruleName,
+				ID:           ruleID,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+				Status:    re.EnabledStatus,
+				CreatedBy: userID,
+				DomainID:  domainID,
+			},
+			err: nil,
+		},
+		{
+			desc: "view rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:  ruleID,
+			err: svcerr.ErrViewEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("ViewRule", mock.Anything, mock.Anything).Return(tc.res, tc.err)
+			res, err := svc.ViewRule(context.Background(), tc.session, tc.id)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.Equal(t, tc.res, res)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestUpdateRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+
+	newName := namegen.Generate()
+	now := time.Now().Add(time.Hour)
+	cases := []struct {
+		desc    string
+		session authn.Session
+		rule    re.Rule
+		res     re.Rule
+		err     error
+	}{
+		{
+			desc: "update rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			rule: re.Rule{
+				Name:         newName,
+				ID:           ruleID,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+				Status:    re.EnabledStatus,
+				CreatedBy: userID,
+				DomainID:  domainID,
+			},
+			res: re.Rule{
+				Name:         newName,
+				ID:           ruleID,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+				Status:    re.EnabledStatus,
+				CreatedBy: userID,
+				DomainID:  domainID,
+				UpdatedAt: now,
+				UpdatedBy: userID,
+			},
+			err: nil,
+		},
+		{
+			desc: "update rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			rule: re.Rule{
+				Name:         ruleName,
+				ID:           ruleID,
+				InputChannel: inputChannel,
+				Schedule: re.Schedule{
+					Recurring:       re.Daily,
+					RecurringPeriod: 1,
+					Time:            now,
+				},
+				Status:    re.EnabledStatus,
+				CreatedBy: userID,
+				DomainID:  domainID,
+			},
+			err: svcerr.ErrUpdateEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("UpdateRule", mock.Anything, mock.Anything).Return(tc.res, tc.err)
+			res, err := svc.UpdateRule(context.Background(), tc.session, tc.rule)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.Equal(t, tc.res, res)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestListRules(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+	numRules := 50
+	now := time.Now().Add(time.Hour)
+	var rules []re.Rule
+	for i := 0; i < numRules; i++ {
+		r := re.Rule{
+			ID:        testsutil.GenerateUUID(t),
+			Name:      namegen.Generate(),
+			DomainID:  domainID,
+			Status:    re.EnabledStatus,
+			CreatedAt: now,
+			CreatedBy: userID,
+			Schedule: re.Schedule{
+				Recurring:       re.Daily,
+				Time:            now.Add(1 * time.Hour),
+				RecurringPeriod: 1,
+				StartDateTime:   now.Add(-1 * time.Hour),
+			},
+		}
+		rules = append(rules, r)
+	}
+
+	cases := []struct {
+		desc     string
+		session  authn.Session
+		pageMeta re.PageMeta
+		res      re.Page
+		err      error
+	}{
+		{
+			desc: "list rules successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			pageMeta: re.PageMeta{},
+			res: re.Page{
+				PageMeta: re.PageMeta{
+					Total:  uint64(numRules),
+					Offset: 0,
+					Limit:  10,
+				},
+				Rules: rules[0:10],
+			},
+			err: nil,
+		},
+		{
+			desc: "list rules successfully with limit",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			pageMeta: re.PageMeta{
+				Limit: 100,
+			},
+			res: re.Page{
+				PageMeta: re.PageMeta{
+					Total:  uint64(numRules),
+					Offset: 0,
+					Limit:  100,
+				},
+				Rules: rules[0:numRules],
+			},
+			err: nil,
+		},
+		{
+			desc: "list rules successfully with offset",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			pageMeta: re.PageMeta{
+				Offset: 20,
+				Limit:  10,
+			},
+			res: re.Page{
+				PageMeta: re.PageMeta{
+					Total:  uint64(numRules),
+					Offset: 20,
+					Limit:  10,
+				},
+				Rules: rules[20:30],
+			},
+			err: nil,
+		},
+		{
+			desc: "list rules with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			pageMeta: re.PageMeta{},
+			err:      svcerr.ErrViewEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("ListRules", mock.Anything, mock.Anything).Return(tc.res, tc.err)
+			res, err := svc.ListRules(context.Background(), tc.session, tc.pageMeta)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.Equal(t, tc.res, res)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestRemoveRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+
+	cases := []struct {
+		desc    string
+		session authn.Session
+		id      string
+		err     error
+	}{
+		{
+			desc: "remove rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:  ruleID,
+			err: nil,
+		},
+		{
+			desc: "remove rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:  ruleID,
+			err: svcerr.ErrRemoveEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("RemoveRule", mock.Anything, mock.Anything).Return(tc.err)
+			err := svc.RemoveRule(context.Background(), tc.session, tc.id)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestEnableRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+
+	cases := []struct {
+		desc    string
+		session authn.Session
+		id      string
+		status  re.Status
+		res     re.Rule
+		err     error
+	}{
+		{
+			desc: "enable rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:     ruleID,
+			status: re.EnabledStatus,
+			res: re.Rule{
+				ID:           ruleID,
+				Name:         ruleName,
+				DomainID:     domainID,
+				InputChannel: inputChannel,
+				Status:       re.EnabledStatus,
+				Schedule:     schedule,
+			},
+			err: nil,
+		},
+		{
+			desc: "enable rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:     ruleID,
+			status: re.EnabledStatus,
+			err:    svcerr.ErrUpdateEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("UpdateRuleStatus", context.Background(), tc.id, tc.status).Return(tc.res, tc.err)
+			res, err := svc.EnableRule(context.Background(), tc.session, tc.id)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.Equal(t, tc.res, res)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestDisableRule(t *testing.T) {
+	svc, repo, _, _ := newService(t)
+
+	cases := []struct {
+		desc    string
+		session authn.Session
+		id      string
+		status  re.Status
+		res     re.Rule
+		err     error
+	}{
+		{
+			desc: "disable rule successfully",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:     ruleID,
+			status: re.DisabledStatus,
+			res: re.Rule{
+				ID:           ruleID,
+				Name:         ruleName,
+				DomainID:     domainID,
+				InputChannel: inputChannel,
+				Status:       re.DisabledStatus,
+				Schedule:     schedule,
+			},
+			err: nil,
+		},
+		{
+			desc: "disable rule with failed repo",
+			session: authn.Session{
+				UserID:   userID,
+				DomainID: domainID,
+			},
+			id:     ruleID,
+			status: re.DisabledStatus,
+			err:    svcerr.ErrUpdateEntity,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoCall := repo.On("UpdateRuleStatus", mock.Anything, tc.id, tc.status).Return(tc.res, tc.err)
+			res, err := svc.DisableRule(context.Background(), tc.session, tc.id)
+
+			assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.err, err))
+			if err == nil {
+				assert.Equal(t, tc.res, res)
+			}
+			defer repoCall.Unset()
+		})
+	}
+}
+
+func TestConsumeAsync(t *testing.T) {
+	svc, repo, pubmocks, _ := newService(t)
+	now := time.Now()
+
+	cases := []struct {
+		desc       string
+		message    any
+		pageMeta   re.PageMeta
+		page       re.Page
+		listErr    error
+		publishErr error
+		expectErr  bool
+	}{
+		{
+			desc: "consume message with empty rules",
+			message: &messaging.Message{
+				Channel: inputChannel,
+				Created: now.Unix(),
+			},
+			pageMeta: re.PageMeta{
+				InputChannel: inputChannel,
+				Status:       re.EnabledStatus,
+			},
+			page: re.Page{
+				Rules: []re.Rule{},
+			},
+			listErr: nil,
+		},
+		{
+			desc: "consume message with rules",
+			message: &messaging.Message{
+				Channel: inputChannel,
+				Created: now.Unix(),
+			},
+			pageMeta: re.PageMeta{
+				InputChannel: inputChannel,
+				Status:       re.EnabledStatus,
+			},
+			page: re.Page{
+				Rules: []re.Rule{
+					{
+						ID:           testsutil.GenerateUUID(t),
+						Name:         namegen.Generate(),
+						InputChannel: inputChannel,
+						Status:       re.EnabledStatus,
+						Logic: re.Script{
+							Type: re.ScriptType(0),
+						},
+						OutputChannel: "output.channel",
+						Schedule:      schedule,
+					},
+				},
+			},
+			listErr: nil,
+		},
+		{
+			desc:    "consume message with unsupported message type",
+			message: "unsupported message type",
+			pageMeta: re.PageMeta{
+				InputChannel: inputChannel,
+				Status:       re.EnabledStatus,
+			},
+			page: re.Page{},
+		},
+		{
+			desc:    "consume json message",
+			message: mgjson.Message{},
+			pageMeta: re.PageMeta{
+				InputChannel: inputChannel,
+				Status:       re.EnabledStatus,
+			},
+			page:    re.Page{},
+			listErr: nil,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+			defer cancel()
+
+			var err error
+
+			repoCall := repo.On("ListRules", mock.Anything, tc.pageMeta).Return(tc.page, tc.listErr).Run(func(args mock.Arguments) {
+				if tc.listErr != nil {
+					err = tc.listErr
+				}
+			})
+			repoCall1 := pubmocks.On("Publish", mock.Anything, mock.Anything, mock.Anything).Return(tc.publishErr)
+
+			svc.ConsumeAsync(ctx, tc.message)
+
+			assert.True(t, errors.Contains(err, tc.listErr), fmt.Sprintf("%s: expected %s got %s\n", tc.desc, tc.listErr, err))
+
+			repoCall.Unset()
+			repoCall1.Unset()
+		})
+	}
 }
 
 func TestStartScheduler(t *testing.T) {
 	now := time.Now().Truncate(time.Minute)
-	svc, repo, ticker := newService(t)
+	svc, repo, _, ticker := newService(t)
+
+	noRecurringPeriod := re.Rule{
+		ID:           testsutil.GenerateUUID(t),
+		Name:         namegen.Generate(),
+		InputChannel: inputChannel,
+		Status:       re.EnabledStatus,
+		Schedule: re.Schedule{
+			StartDateTime:   time.Now().Add(-time.Hour),
+			Recurring:       re.None,
+			RecurringPeriod: 0,
+			Time:            time.Now().Add(-time.Hour),
+		},
+	}
+
+	weeklyRule := re.Rule{
+		ID:           testsutil.GenerateUUID(t),
+		Name:         namegen.Generate(),
+		InputChannel: inputChannel,
+		Status:       re.EnabledStatus,
+		Schedule: re.Schedule{
+			StartDateTime:   time.Now().Add(-time.Hour),
+			Recurring:       re.Weekly,
+			RecurringPeriod: 1,
+			Time:            time.Now().Add(-time.Hour),
+		},
+	}
+
+	monthlyRule := re.Rule{
+		ID:           testsutil.GenerateUUID(t),
+		Name:         namegen.Generate(),
+		InputChannel: inputChannel,
+		Status:       re.EnabledStatus,
+		Schedule: re.Schedule{
+			StartDateTime:   time.Now().Add(-time.Hour),
+			Recurring:       re.Monthly,
+			RecurringPeriod: 1,
+			Time:            time.Now().Add(-time.Hour),
+		},
+	}
+
+	pastRule := re.Rule{
+		ID:           testsutil.GenerateUUID(t),
+		Name:         namegen.Generate(),
+		InputChannel: inputChannel,
+		Status:       re.EnabledStatus,
+		Schedule: re.Schedule{
+			StartDateTime:   time.Now().Add(-time.Hour),
+			Recurring:       re.None,
+			RecurringPeriod: 1,
+			Time:            time.Now().Add(-time.Hour),
+		},
+	}
 
 	cases := []struct {
 		desc     string
@@ -142,6 +783,62 @@ func TestStartScheduler(t *testing.T) {
 				return context.WithCancel(context.Background())
 			},
 		},
+		{
+			desc: "start scheduler successfully with no recurring period",
+			err:  context.Canceled,
+			pageMeta: re.PageMeta{
+				Status:          re.EnabledStatus,
+				ScheduledBefore: &now,
+			},
+			page: re.Page{
+				Rules: []re.Rule{noRecurringPeriod},
+			},
+			setupCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(context.Background())
+			},
+		},
+		{
+			desc: "start scheduler successfully with weekly schedule",
+			err:  context.Canceled,
+			pageMeta: re.PageMeta{
+				Status:          re.EnabledStatus,
+				ScheduledBefore: &now,
+			},
+			page: re.Page{
+				Rules: []re.Rule{weeklyRule},
+			},
+			setupCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(context.Background())
+			},
+		},
+		{
+			desc: "start scheduler successfully with monthly schedule",
+			err:  context.Canceled,
+			pageMeta: re.PageMeta{
+				Status:          re.EnabledStatus,
+				ScheduledBefore: &now,
+			},
+			page: re.Page{
+				Rules: []re.Rule{monthlyRule},
+			},
+			setupCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(context.Background())
+			},
+		},
+		{
+			desc: "start scheduler successfully processes rules with past schedule",
+			err:  context.Canceled,
+			pageMeta: re.PageMeta{
+				Status:          re.EnabledStatus,
+				ScheduledBefore: &now,
+			},
+			page: re.Page{
+				Rules: []re.Rule{pastRule},
+			},
+			setupCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(context.Background())
+			},
+		},
 	}
 
 	for _, tc := range cases {
@@ -161,20 +858,16 @@ func TestStartScheduler(t *testing.T) {
 			switch tc.desc {
 			case "start scheduler with canceled context":
 				cancel()
-			case "start scheduler successfully processes rules":
-				tickChan <- time.Now()
-				time.Sleep(100 * time.Millisecond)
-				cancel()
-			case "start scheduler with rule to be run in the future":
-				tickChan <- time.Now()
-				time.Sleep(100 * time.Millisecond)
-				cancel()
 			case "start scheduler with list error":
 				tickChan <- time.Now()
 				time.Sleep(100 * time.Millisecond)
 				if err := svc.Errors(); err != nil {
 					cancel()
 				}
+			default:
+				tickChan <- time.Now()
+				time.Sleep(100 * time.Millisecond)
+				cancel()
 			}
 
 			err := <-errc