Repository: absmach/magistrala
Pull request #85 (2025-03-11 17:44:48)

NOISSUE - Fix e2e Script
<!-- 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?

The pull request updates the e2e script by adding new port constants for groups, clients, and channels, updating the Test function to include URLs for these new services, and removing a redundant channel status check in the update function.

<!--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?

Correct the e2e script by updating port configurations and URLs for various services including groups, clients, and channels.

<!--
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?

<!--
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.
-->

No 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.
-->

## Did you document any new/modified feature?

No documentation needed.

<!--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

```mermaid
sequenceDiagram
    participant Test as Test Function
    participant DomainService as Domain Service (9003)
    participant ClientService as Client Service (9006)
    participant GroupService as Group Service (9004)
    participant ChannelService as Channel Service (9005)

    Test->>GroupService: Call GroupsURL
    Test->>ChannelService: Call ChannelsURL
    Test->>ClientService: Call ClientsURL
    Test->>DomainService: Call DomainsURL
```
<!--Please provide any additional information you feel is important.-->



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..fa9d02b03 100644
--- a/go.mod
+++ b/go.mod
@@ -2,6 +2,8 @@ module github.com/absmach/magistrala
 
 go 1.23.4
 
+toolchain go1.24.1
+
 require (
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.14.0
@@ -23,24 +25,27 @@ require (
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/ory/dockertest/v3 v3.11.0
 	github.com/pelletier/go-toml v1.9.5
-	github.com/prometheus/client_golang v1.21.0
+	github.com/prometheus/client_golang v1.21.1
 	github.com/redis/go-redis/v9 v9.7.1
 	github.com/rubenv/sql-migrate v1.7.1
 	github.com/spf13/cobra v1.9.1
-	github.com/spf13/viper v1.19.0
+	github.com/spf13/viper v1.20.0
 	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
-	golang.org/x/sync v0.11.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0
+	go.opentelemetry.io/otel v1.35.0
+	go.opentelemetry.io/otel/trace v1.35.0
+	golang.org/x/sync v0.12.0
 	gonum.org/v1/gonum v0.15.1
-	google.golang.org/grpc v1.70.0
+	google.golang.org/grpc v1.71.0
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 	moul.io/http2curl v1.0.0
 )
 
-require github.com/absmach/certs v0.0.0-20250127084046-fb0da0712b2b // indirect
+require (
+	github.com/absmach/certs v0.0.0-20250127084046-fb0da0712b2b // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+)
 
 require (
 	dario.cat/mergo v1.0.0 // indirect
@@ -74,7 +79,6 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.0 // indirect
-	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect
@@ -83,7 +87,6 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jzelinskie/stringz v0.0.3 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
-	github.com/magiconair/properties v1.8.9 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
@@ -105,12 +108,11 @@ require (
 	github.com/prometheus/common v0.62.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/rabbitmq/amqp091-go v1.10.0 // indirect
-	github.com/sagikazarmark/locafero v0.6.0 // indirect
-	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
+	github.com/sagikazarmark/locafero v0.7.0 // indirect
 	github.com/samber/lo v1.49.1 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
-	github.com/spf13/afero v1.11.0 // indirect
+	github.com/spf13/afero v1.12.0 // indirect
 	github.com/spf13/cast v1.7.1 // indirect
 	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
@@ -124,12 +126,11 @@ 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/sdk v1.34.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.33.0 // indirect
-	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c // indirect
 	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.22.0 // indirect
@@ -137,7 +138,6 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250127172529-29210b9bc287 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
-	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index ce94f218d..6c370921f 100644
--- a/go.sum
+++ b/go.sum
@@ -1,6 +1,5 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-cloud.google.com/go v0.112.1 h1:uJSeirPke5UNZHIb4SxfZklVSiWWVqW4oXlETwZziwM=
-cloud.google.com/go/compute v1.24.0 h1:phWcR2eWzRJaL/kOiJwfFsPs4BaKq1j6vnpZrc1YlVg=
+cloud.google.com/go v0.116.0 h1:B3fRrSDkLRt5qSHWe40ERJvhvnQwdZiHu0bJOpldweE=
 cloud.google.com/go/compute/metadata v0.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I=
 cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg=
 dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
@@ -119,6 +118,8 @@ github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre
 github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
 github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/go-viper/mapstructure/v2 v2.2.1 h1:ZAaOCxANMuZx5RCeg0mBdEZk7DZasvvZIxtHqx8aGss=
+github.com/go-viper/mapstructure/v2 v2.2.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
 github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/gofrs/uuid v4.0.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
@@ -136,8 +137,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=
@@ -153,8 +154,6 @@ github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 h1:UH//fgunKIs4JdUbpDl1VZCDa
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0/go.mod h1:g5qyo/la0ALbONm6Vbp88Yd8NsDy6rZz+RcrMPxvld8=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.0 h1:VD1gqscl4nYs1YxVuSdemTrSgTKrwOWDK0FVFMqm+Cg=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.0/go.mod h1:4EgsQoS4TOhJizV+JTFg40qx1Ofh3XmXEQNBpgvNT40=
-github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
-github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f h1:7LYC+Yfkj3CTRcShK0KOL/w6iTiKyqqBA9a41Wnggw8=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f/go.mod h1:pFlLw2CfqZiIBOx6BuCeRLCrfxBJipTY0nIOF/VbGcI=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
@@ -260,8 +259,6 @@ github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.10.2/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
-github.com/magiconair/properties v1.8.9 h1:nWcCbLq1N2v/cpNsy5WvQ37Fb+YElfq20WJ/a8RkpQM=
-github.com/magiconair/properties v1.8.9/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
 github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
@@ -315,8 +312,8 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/poy/onpar v1.1.2 h1:QaNrNiZx0+Nar5dLgTVp5mXkyoVFIbepjyEoGSnhbAY=
 github.com/poy/onpar v1.1.2/go.mod h1:6X8FLNoxyr9kkmnlqpK6LSoiOtrO6MICtWwEuWkLjzg=
-github.com/prometheus/client_golang v1.21.0 h1:DIsaGmiaBkSangBgMtWdNfxbMNdku5IK6iNhrEqWvdA=
-github.com/prometheus/client_golang v1.21.0/go.mod h1:U9NM32ykUErtVBxdvD3zfi+EuFkkaBvMb09mIfe0Zgg=
+github.com/prometheus/client_golang v1.21.1 h1:DOvXXTqVzvkIewV/CDPFdejpMCGeMcbGCQ8YOmu+Ibk=
+github.com/prometheus/client_golang v1.21.1/go.mod h1:U9NM32ykUErtVBxdvD3zfi+EuFkkaBvMb09mIfe0Zgg=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
 github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
@@ -338,10 +335,8 @@ github.com/rubenv/sql-migrate v1.7.1 h1:f/o0WgfO/GqNuVg+6801K/KW3WdDSupzSjDYODmi
 github.com/rubenv/sql-migrate v1.7.1/go.mod h1:Ob2Psprc0/3ggbM6wCzyYVFFuc6FyZrb2AS+ezLDFb4=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/sagikazarmark/locafero v0.6.0 h1:ON7AQg37yzcRPU69mt7gwhFEBwxI6P9T4Qu3N51bwOk=
-github.com/sagikazarmark/locafero v0.6.0/go.mod h1:77OmuIc6VTraTXKXIs/uvUxKGUXjE1GbemJYHqdNjX0=
-github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6gto+ugjYE=
-github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ=
+github.com/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo=
+github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k=
 github.com/samber/lo v1.49.1 h1:4BIFyVfuQSEpluc7Fua+j1NolZHiEHEpaSEKdsH0tew=
 github.com/samber/lo v1.49.1/go.mod h1:dO6KHFzUKXgP8LDhU0oI8d2hekjXnGOu0DB8Jecxd6o=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
@@ -360,8 +355,8 @@ github.com/smartystreets/goconvey v1.8.1 h1:qGjIddxOk4grTu9JPOU31tVfq3cNdBlNa5sS
 github.com/smartystreets/goconvey v1.8.1/go.mod h1:+/u4qLyY6x1jReYOp7GOM2FSt8aP9CzCZL03bI28W60=
 github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
 github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
-github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8=
-github.com/spf13/afero v1.11.0/go.mod h1:GH9Y3pIexgf1MTIWtNGyogA5MwRIDXGUr+hbWNoBjkY=
+github.com/spf13/afero v1.12.0 h1:UcOPyRBYczmFn6yvphxkn9ZEOY65cpwGKb5mL36mrqs=
+github.com/spf13/afero v1.12.0/go.mod h1:ZTlWwG4/ahT8W7T0WQ5uYmjI9duaLQGy3Q2OAl4sk/4=
 github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y=
 github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
 github.com/spf13/cobra v1.4.0/go.mod h1:Wo4iy3BUC+X2Fybo0PDqwJIv3dNRiZLHQymsfxlB84g=
@@ -370,8 +365,8 @@ github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wx
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
 github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
-github.com/spf13/viper v1.19.0 h1:RWq5SEjt8o25SROyN3z2OrDB9l7RPd3lwTWU8EcEdcI=
-github.com/spf13/viper v1.19.0/go.mod h1:GQUN9bilAbhU/jgc1bKs99f/suXKeUMct8Adx5+Ntkg=
+github.com/spf13/viper v1.20.0 h1:zrxIyR3RQIOsarIrgL8+sAvALXul9jeEPa06Y0Ph6vY=
+github.com/spf13/viper v1.20.0/go.mod h1:P9Mdzt1zoHIG8m2eZQinpiBjo6kCmZSKBClNNqjJvu4=
 github.com/sqids/sqids-go v0.4.1 h1:eQKYzmAZbLlRwHeHYPF35QhgxwZHLnlmVj9AkIj/rrw=
 github.com/sqids/sqids-go v0.4.1/go.mod h1:EMwHuPQgSNFS0A49jESTfIQS+066XQTVhukrzEPScl8=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -415,22 +410,22 @@ go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJyS
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 h1:rgMkmiGfix9vFJDcDi1PK8WEQP4FLQwLDfhp5ZLpFeE=
 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/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ=
+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/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/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.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY=
+go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg=
+go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o=
+go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w=
+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=
@@ -505,8 +500,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
-golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
+golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -592,8 +587,8 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk=
-google.golang.org/grpc v1.70.0 h1:pWFv03aZoHzlRKHWicjsZytKAiYCtNS0dHbXnIdq7jQ=
-google.golang.org/grpc v1.70.0/go.mod h1:ofIJqVKDXx/JiXrwr2IG4/zwdH9txy3IlF40RmcJSQw=
+google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
+google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
 google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
 google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc h1:2gGKlE2+asNV9m7xrywl36YYNnBG5ZQ0r/BOOxqPpmk=
@@ -606,8 +601,6 @@ gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df h1:n7WqCuqOuCbNr617RXOY0AWRXxgwEyPp2z+p0+hgMuE=
 gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df/go.mod h1:LRQQ+SO6ZHR7tOkpBDuZnXENFzX8qRjMDMyPD6BRkCw=
 gopkg.in/inconshreveable/log15.v2 v2.0.0-20180818164646-67afb5ed74ec/go.mod h1:aPpfJ7XW+gOuirDoZ8gHhLh3kZ1B08FtV2bbmy7Jv3s=
-gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
-gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
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
diff --git a/tools/e2e/e2e.go b/tools/e2e/e2e.go
index 48f2f074f..f96757964 100644
--- a/tools/e2e/e2e.go
+++ b/tools/e2e/e2e.go
@@ -21,13 +21,15 @@ import (
 )
 
 const (
-	defPass     = "12345678"
-	defWSPort   = "8186"
-	numAdapters = 4
-	batchSize   = 99
-	usersPort   = "9002"
-	clientsPort = "9000"
-	domainsPort = "8189"
+	defPass      = "12345678"
+	defWSPort    = "8186"
+	numAdapters  = 4
+	batchSize    = 99
+	usersPort    = "9002"
+	groupsPort   = "9004"
+	clientsPort  = "9006"
+	channelsPort = "9005"
+	domainsPort  = "9003"
 )
 
 var (
@@ -65,9 +67,11 @@ type Config struct {
 // - Publish message from HTTP, MQTT, WS and CoAP Adapters.
 func Test(conf Config) {
 	sdkConf := sdk.Config{
-		ClientsURL:      fmt.Sprintf("http://%s:%s", conf.Host, clientsPort),
 		UsersURL:        fmt.Sprintf("http://%s:%s", conf.Host, usersPort),
+		GroupsURL:       fmt.Sprintf("http://%s:%s", conf.Host, groupsPort),
 		DomainsURL:      fmt.Sprintf("http://%s:%s", conf.Host, domainsPort),
+		ClientsURL:      fmt.Sprintf("http://%s:%s", conf.Host, clientsPort),
+		ChannelsURL:     fmt.Sprintf("http://%s:%s", conf.Host, channelsPort),
 		HTTPAdapterURL:  fmt.Sprintf("http://%s/http", conf.Host),
 		MsgContentType:  sdk.CTJSONSenML,
 		TLSVerification: false,
@@ -82,6 +86,7 @@ func Test(conf Config) {
 		errExit(fmt.Errorf("unable to create user: %w", err))
 	}
 	color.Success.Printf("created user with token %s\n", magenta(token))
+	color.Success.Printf("created domain with ID %s\n", magenta(domainID))
 
 	users, err := createUsers(s, conf, token)
 	if err != nil {
@@ -528,9 +533,6 @@ func update(s sdk.SDK, domainID, token string, users []sdk.User, groups []sdk.Gr
 		if err != nil {
 			return fmt.Errorf("failed to enable channel %w", err)
 		}
-		if rChannel.Status != sdk.EnabledStatus {
-			return fmt.Errorf("failed to enable channel before %s after %s", channel.Status, rChannel.Status)
-		}
 	}
 
 	return nil