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

Bump github.com/prometheus/client_golang from 1.21.0 to 1.21.1
Bumps [github.com/prometheus/client_golang](https://github.com/prometheus/client_golang) from 1.21.0 to 1.21.1.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/prometheus/client_golang/releases">github.com/prometheus/client_golang's releases</a>.</em></p>
<blockquote>
<h2>v1.21.1 / 2025-03-04</h2>
<p>This release addresses a performance regression introduced in <a href="https://redirect.github.com/prometheus/client_golang/issues/1661">#1661</a> -- thanks to all who <a href="https://redirect.github.com/prometheus/client_golang/issues/1748">reported this quickly</a>:
<a href="https://github.com/chlunde"><code>@​chlunde</code></a>, <a href="https://github.com/dethi"><code>@​dethi</code></a>, <a href="https://github.com/aaronbee"><code>@​aaronbee</code></a> <a href="https://github.com/tsuna"><code>@​tsuna</code></a> <a href="https://github.com/kakkoyun"><code>@​kakkoyun</code></a>  💪🏽. This patch release also fixes the iOS build.</p>
<p>We will be hardening the release process even further (<a href="https://redirect.github.com/prometheus/client_golang/issues/1759">#1759</a>, <a href="https://redirect.github.com/prometheus/client_golang/issues/1761">#1761</a>) to prevent this in future, sorry for the inconvenience!</p>
<p>The high concurrency optimization is planned to be eventually reintroduced, however in a much safer manner, potentially in a separate API.</p>
<ul>
<li>[BUGFIX] prometheus: Revert of <code>Inc</code>, <code>Add</code> and <code>Observe</code> cumulative metric CAS optimizations (<a href="https://redirect.github.com/prometheus/client_golang/issues/1661">#1661</a>), causing regressions on low concurrency cases <a href="https://redirect.github.com/prometheus/client_golang/issues/1757">#1757</a></li>
<li>[BUGFIX] prometheus: Fix GOOS=ios build, broken due to process_collector_* wrong build tags. <a href="https://redirect.github.com/prometheus/client_golang/issues/1758">#1758</a></li>
</ul>
<!-- raw HTML omitted -->
<ul>
<li>Revert &quot;exponential backoff for CAS operations on floats&quot; and cut 1.21.1 by <a href="https://github.com/bwplotka"><code>@​bwplotka</code></a> in <a href="https://redirect.github.com/prometheus/client_golang/pull/1757">prometheus/client_golang#1757</a></li>
<li>Fix ios build for 1.21.1 by <a href="https://github.com/bwplotka"><code>@​bwplotka</code></a> in <a href="https://redirect.github.com/prometheus/client_golang/pull/1758">prometheus/client_golang#1758</a></li>
</ul>
<!-- raw HTML omitted -->
<p><strong>Full Changelog</strong>: <a href="https://github.com/prometheus/client_golang/compare/v1.21.0...v1.21.1">https://github.com/prometheus/client_golang/compare/v1.21.0...v1.21.1</a></p>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/prometheus/client_golang/blob/main/CHANGELOG.md">github.com/prometheus/client_golang's changelog</a>.</em></p>
<blockquote>
<h2>1.21.1 / 2025-03-04</h2>
<ul>
<li>[BUGFIX] prometheus: Revert of <code>Inc</code>, <code>Add</code> and <code>Observe</code> cumulative metric CAS optimizations (<a href="https://redirect.github.com/prometheus/client_golang/issues/1661">#1661</a>), causing regressions on low contention cases.</li>
<li>[BUGFIX] prometheus: Fix GOOS=ios build, broken due to process_collector_* wrong build tags.</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/prometheus/client_golang/commit/8a42da3e4bfdc7ea408fcd886064dffc4daffb22"><code>8a42da3</code></a> Fix ios build. (<a href="https://redirect.github.com/prometheus/client_golang/issues/1758">#1758</a>)</li>
<li><a href="https://github.com/prometheus/client_golang/commit/40c62f7a974e27ba11164f7138d60fd7a3b0c326"><code>40c62f7</code></a> Merge pull request <a href="https://redirect.github.com/prometheus/client_golang/issues/1757">#1757</a> from prometheus/revert-121cas</li>
<li><a href="https://github.com/prometheus/client_golang/commit/689f5900164029a73c9a2be6068e1db615107c1f"><code>689f590</code></a> Cut 1.21.1</li>
<li><a href="https://github.com/prometheus/client_golang/commit/9e567a7b4fe821af4fad882552ec06713fb8d291"><code>9e567a7</code></a> Revert &quot;Add: exponential backoff for CAS operations on floats (<a href="https://redirect.github.com/prometheus/client_golang/issues/1661">#1661</a>)&quot;</li>
<li>See full diff in <a href="https://github.com/prometheus/client_golang/compare/v1.21.0...v1.21.1">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=github.com/prometheus/client_golang&package-manager=go_modules&previous-version=1.21.0&new-version=1.21.1)](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>


Patch:
diff --git a/bootstrap/postgres/configs.go b/bootstrap/postgres/configs.go
index 73b3130b1..3b6a5aa86 100644
--- a/bootstrap/postgres/configs.go
+++ b/bootstrap/postgres/configs.go
@@ -100,15 +100,11 @@ func (cr configRepository) RetrieveByID(ctx context.Context, domainID, id string
 	}
 	row, err := cr.db.NamedQueryContext(ctx, q, dbcfg)
 	if err != nil {
-		if err == sql.ErrNoRows {
-			return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, err)
-		}
-
 		return bootstrap.Config{}, errors.Wrap(repoerr.ErrViewEntity, err)
 	}
 
-	if ok := row.Next(); !ok {
-		return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, row.Err())
+	if !row.Next() {
+		return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, sql.ErrNoRows)
 	}
 
 	if err := row.StructScan(&dbcfg); err != nil {
@@ -205,14 +201,11 @@ func (cr configRepository) RetrieveByExternalID(ctx context.Context, externalID
 
 	row, err := cr.db.NamedQueryContext(ctx, q, dbcfg)
 	if err != nil {
-		if err == sql.ErrNoRows {
-			return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, err)
-		}
 		return bootstrap.Config{}, errors.Wrap(repoerr.ErrViewEntity, err)
 	}
 
-	if ok := row.Next(); !ok {
-		return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, row.Err())
+	if !row.Next() {
+		return bootstrap.Config{}, errors.Wrap(repoerr.ErrNotFound, sql.ErrNoRows)
 	}
 
 	if err := row.StructScan(&dbcfg); err != nil {
diff --git a/cli/config.go b/cli/config.go
index e4e513d83..3e7b1df68 100644
--- a/cli/config.go
+++ b/cli/config.go
@@ -27,7 +27,6 @@ const (
 	defBootstrapURL    string = defURL + ":9013"
 	defDomainsURL      string = defURL + ":9003"
 	defCertsURL        string = defURL + ":9019"
-	defInvitationsURL  string = defURL + ":9020"
 	defHTTPURL         string = defURL + ":8008"
 	defJournalURL      string = defURL + ":9021"
 	defTLSVerification bool   = false
@@ -47,7 +46,6 @@ type remotes struct {
 	HTTPAdapterURL  string `toml:"http_adapter_url"`
 	BootstrapURL    string `toml:"bootstrap_url"`
 	CertsURL        string `toml:"certs_url"`
-	InvitationsURL  string `toml:"invitations_url"`
 	JournalURL      string `toml:"journal_url"`
 	HostURL         string `toml:"host_url"`
 	TLSVerification bool   `toml:"tls_verification"`
@@ -120,7 +118,6 @@ func ParseConfig(sdkConf mgsdk.Config) (mgsdk.Config, error) {
 				HTTPAdapterURL:  defHTTPURL,
 				BootstrapURL:    defBootstrapURL,
 				CertsURL:        defCertsURL,
-				InvitationsURL:  defInvitationsURL,
 				JournalURL:      defJournalURL,
 				HostURL:         defURL,
 				TLSVerification: defTLSVerification,
@@ -205,10 +202,6 @@ func ParseConfig(sdkConf mgsdk.Config) (mgsdk.Config, error) {
 		sdkConf.CertsURL = config.Remotes.CertsURL
 	}
 
-	if sdkConf.InvitationsURL == "" && config.Remotes.InvitationsURL != "" {
-		sdkConf.InvitationsURL = config.Remotes.InvitationsURL
-	}
-
 	if sdkConf.JournalURL == "" && config.Remotes.JournalURL != "" {
 		sdkConf.JournalURL = config.Remotes.JournalURL
 	}
diff --git a/cmd/cli/main.go b/cmd/cli/main.go
index e8f1c674f..5d3f40256 100644
--- a/cmd/cli/main.go
+++ b/cmd/cli/main.go
@@ -140,14 +140,6 @@ func main() {
 		"Reader URL",
 	)
 
-	rootCmd.PersistentFlags().StringVarP(
-		&mgsdkConf.InvitationsURL,
-		"invitations-url",
-		"v",
-		mgsdkConf.InvitationsURL,
-		"Inivitations URL",
-	)
-
 	rootCmd.PersistentFlags().StringVarP(
 		&mgsdkConf.JournalURL,
 		"journal-url",
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 9dd792169..92c72a31d 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.23.4
 require (
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.14.0
-	github.com/absmach/supermq v0.16.1-0.20250113091433-3a11b54394b2
+	github.com/absmach/supermq v0.16.1-0.20250220214139-eaaba5beebee
 	github.com/authzed/authzed-go v1.3.0
 	github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b
 	github.com/caarlos0/env/v11 v11.3.1
@@ -23,7 +23,7 @@ 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
@@ -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
@@ -40,7 +40,7 @@ require (
 	moul.io/http2curl v1.0.0
 )
 
-require github.com/absmach/certs v0.0.0-20241209153600-91270de67b5a // indirect
+require github.com/absmach/certs v0.0.0-20250127084046-fb0da0712b2b // indirect
 
 require (
 	dario.cat/mergo v1.0.0 // indirect
@@ -84,13 +84,13 @@ require (
 	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.13 // 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
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/nats-io/nats.go v1.38.0 // indirect
+	github.com/nats-io/nats.go v1.39.0 // indirect
 	github.com/nats-io/nkeys v0.4.9 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oklog/ulid/v2 v2.1.0 // indirect
@@ -107,7 +107,7 @@ require (
 	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/samber/lo v1.49.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
@@ -121,21 +121,21 @@ require (
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.58.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.33.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.33.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.4.0 // indirect
+	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.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
-	golang.org/x/crypto v0.32.0 // indirect
-	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 // 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.29.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20250115164207-1a7da9e5054f // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
-	google.golang.org/protobuf v1.36.4 // indirect
+	golang.org/x/sys v0.30.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250127172529-29210b9bc287 // indirect
+	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
diff --git a/go.sum b/go.sum
index 83d32de98..e807080e5 100644
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,8 @@
 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/compute/metadata v0.5.2 h1:UxK4uu/Tn+I3p2dYWTfiX4wva7aYlKixAHn3fyqngqo=
-cloud.google.com/go/compute/metadata v0.5.2/go.mod h1:C66sj2AluDcIqakBq/M8lw8/ybHgOZqin2obFxa/E5k=
+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=
 dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
@@ -21,14 +21,14 @@ github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrd
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
 github.com/absmach/callhome v0.14.0 h1:zB4tIZJ1YUmZ1VGHFPfMA/Lo6/Mv19y2dvoOiXj2BWs=
 github.com/absmach/callhome v0.14.0/go.mod h1:l12UJOfibK4Muvg/AbupHuquNV9qSz/ROdTEPg7f2Vk=
-github.com/absmach/certs v0.0.0-20241209153600-91270de67b5a h1:gYbkOHsyOALSoR9D+A7fhgqGChMfakhMxy+6y3AGPPE=
-github.com/absmach/certs v0.0.0-20241209153600-91270de67b5a/go.mod h1:g6Kqge7RVxwt+LRxqt+09cqa2SgPAwXvIPoyPsEqZlQ=
+github.com/absmach/certs v0.0.0-20250127084046-fb0da0712b2b h1:EGIqL1bARjRSS7kH98Q5O/g7lZN/Q0KtAVX5mxRcq84=
+github.com/absmach/certs v0.0.0-20250127084046-fb0da0712b2b/go.mod h1:g6Kqge7RVxwt+LRxqt+09cqa2SgPAwXvIPoyPsEqZlQ=
 github.com/absmach/mgate v0.4.5 h1:l6RmrEsR9jxkdb9WHUSecmT0HA41TkZZQVffFfUAIfI=
 github.com/absmach/mgate v0.4.5/go.mod h1:IvRIHZexZPEIAPmmaJF0L5DY2ERjj+GxRGitOW4s6qo=
 github.com/absmach/senml v1.0.6 h1:WPeIl6vQ00k7ghWSZYT/QP0KUxq2+4zQoaC7240pLFk=
 github.com/absmach/senml v1.0.6/go.mod h1:QnJNPy1DJPy0+qUW21PTcH/xoh0LgfYZxTfwriMIvmQ=
-github.com/absmach/supermq v0.16.1-0.20250113091433-3a11b54394b2 h1:mG0ucTmOE41BOzKpZyNkGHbLFangumlZd3TUddJ0J2U=
-github.com/absmach/supermq v0.16.1-0.20250113091433-3a11b54394b2/go.mod h1:DjM4b/VF9m/o/MmQ/WUExbdatIvPnqyREe9zFEbW2YI=
+github.com/absmach/supermq v0.16.1-0.20250220214139-eaaba5beebee h1:V6A4nEZsIY/fqCyN6VNTux4cW82LKVfIAkHk8y3eb6g=
+github.com/absmach/supermq v0.16.1-0.20250220214139-eaaba5beebee/go.mod h1:IOgfP1hPz5g+FqDgjmVW8ChLBqr5T7rLU8lHHRMy8Y8=
 github.com/authzed/authzed-go v1.3.0 h1:jKIMpYDy+6WoOwl32HRURxLZxNGm+I7ObUlTntEPcXA=
 github.com/authzed/authzed-go v1.3.0/go.mod h1:MYkXImtFAxrM/bVZvmC/WO+gZC9RLlvpCM51SLaUZb0=
 github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b h1:wbh8IK+aMLTCey9sZasO7b6BWLAJnHHvb79fvWCXwxw=
@@ -119,8 +119,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/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
-github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+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=
 github.com/gofrs/uuid/v5 v5.3.1 h1:aPx49MwJbekCzOyhZDjJVb0hx3A0KLjlbLx6p2gY0p0=
 github.com/gofrs/uuid/v5 v5.3.1/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8=
@@ -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=
@@ -265,13 +265,12 @@ github.com/magiconair/properties v1.8.9/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3v
 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=
-github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
-github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
+github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
 github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
-github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
@@ -284,8 +283,8 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
 github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
-github.com/nats-io/nats.go v1.38.0 h1:A7P+g7Wjp4/NWqDOOP/K6hfhr54DvdDQUznt5JFg9XA=
-github.com/nats-io/nats.go v1.38.0/go.mod h1:IGUM++TwokGnXPs82/wCuiHS02/aKrdYUQkU8If6yjw=
+github.com/nats-io/nats.go v1.39.0 h1:2/yg2JQjiYYKLwDuBzV0FbB2sIV+eFNkEevlRi4n9lI=
+github.com/nats-io/nats.go v1.39.0/go.mod h1:MgRb8oOdigA6cYpEPhXJuRVH6UE/V4jblJ2jQ27IXYM=
 github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
 github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -316,8 +315,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=
@@ -343,8 +342,8 @@ github.com/sagikazarmark/locafero v0.6.0 h1:ON7AQg37yzcRPU69mt7gwhFEBwxI6P9T4Qu3
 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/samber/lo v1.49.0 h1:AGnTnQrg1jpFuwECPUSoxZCfVH5W22b605kWSry3YxM=
-github.com/samber/lo v1.49.0/go.mod h1:dO6KHFzUKXgP8LDhU0oI8d2hekjXnGOu0DB8Jecxd6o=
+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=
 github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
 github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
@@ -414,26 +413,26 @@ github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7
 github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.58.0 h1:PS8wXpbyaDJQ2VDHHncMe9Vct0Zn1fEjpsjrLxGJoSc=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.58.0/go.mod h1:HDBUsEjOuRC0EzKZ1bSaRGZWUBAzo+MhAcUUORSr4D0=
+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/otel/exporters/otlp/otlptrace v1.33.0 h1:Vh5HayB/0HHfOQA7Ctx69E/Y/DcQSMPpKANYVMQ7fBA=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0/go.mod h1:cpgtDBaqD/6ok/UG0jT15/uKjAY8mRA53diogHBg3UI=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.33.0 h1:wpMfgF8E1rkrT1Z6meFh1NDtownE9Ii3n3X2GJYjsaU=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.33.0/go.mod h1:wAy0T/dUbs468uOlkT31xjvqQgEVXv58BRFWEgn5v/0=
-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.33.0 h1:iax7M131HuAm9QkZotNHEfstof92xM+N8sr3uHXc2IM=
-go.opentelemetry.io/otel/sdk v1.33.0/go.mod h1:A1Q5oi7/9XaMlIWzPSxLRWOI8nG3FnzHJNbiENQuihM=
+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.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/proto/otlp v1.4.0 h1:TA9WRvW6zMwP+Ssb6fLoUIuirti1gGbP28GcKG1jgeg=
-go.opentelemetry.io/proto/otlp v1.4.0/go.mod h1:PPBWZIP98o2ElSqI35IHfu7hIhSwvc5N38Jw8pXuGFY=
+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=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
@@ -465,11 +464,11 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
 golang.org/x/crypto v0.20.0/go.mod h1:Xwo95rrVNIoSMx9wa1JroENMToLWn3RNVrTBpLHgZPQ=
-golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
-golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
+golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus=
+golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 h1:1UoZQm6f0P/ZO0w1Ri+f+ifG/gXhegadRdwBIXEFWDo=
-golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67/go.mod h1:qj5a5QZpwLU2NLQudwIN5koi3beDhSAlJwa67PuM98c=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c h1:KL/ZBHXgKGVmuZBZ01Lt57yE5ws8ZPSkkihmEyq7FXc=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c/go.mod h1:tujkw807nyEEAamNbDrEGzRav+ilXA7PCRAd6xsmwiU=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -497,8 +496,8 @@ golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
 golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
-golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.26.0 h1:afQXWNNaeC4nvZ0Ed9XvCCzXM6UHJG7iCg0W4fPqSBE=
+golang.org/x/oauth2 v0.26.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -528,13 +527,12 @@ golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
-golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -550,11 +548,11 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg=
-golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY=
+golang.org/x/time v0.9.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -585,10 +583,10 @@ google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
 google.golang.org/genproto v0.0.0-20200423170343-7949de9c1215/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
-google.golang.org/genproto/googleapis/api v0.0.0-20250115164207-1a7da9e5054f h1:gap6+3Gk41EItBuyi4XX/bp4oqJ3UwuIMl25yGinuAA=
-google.golang.org/genproto/googleapis/api v0.0.0-20250115164207-1a7da9e5054f/go.mod h1:Ic02D47M+zbarjYYUlK57y316f2MoN0gjAwI3f2S95o=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f h1:OxYkA3wjPsZyBylwymxSHa7ViiW1Sml4ToBrncvFehI=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f/go.mod h1:+2Yz8+CLJbIfL9z73EW45avw8Lmge3xVElCP9zEKi50=
+google.golang.org/genproto/googleapis/api v0.0.0-20250127172529-29210b9bc287 h1:A2ni10G3UlplFrWdCDJTl7D7mJ7GSRm37S+PDimaKRw=
+google.golang.org/genproto/googleapis/api v0.0.0-20250127172529-29210b9bc287/go.mod h1:iYONQfRdizDB8JJBybql13nArx91jcUk7zCXEsOofM4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250127172529-29210b9bc287 h1:J1H9f+LEdWAfHcez/4cvaVBox7cOYT+IU6rgqj5x++8=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250127172529-29210b9bc287/go.mod h1:8BS3B93F/U1juMFq9+EDk+qOT5CO1R9IzXxG3PTqiRk=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
@@ -596,8 +594,8 @@ google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8
 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/protobuf v1.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
-google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+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=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc/go.mod h1:m7x9LTH6d71AHyAX77c9yqWCCa3UKHcVEj9y7hAtKDk=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/pkg/sdk/mocks/sdk.go b/pkg/sdk/mocks/sdk.go
index 205cafb40..03aedc1b1 100644
--- a/pkg/sdk/mocks/sdk.go
+++ b/pkg/sdk/mocks/sdk.go
@@ -4561,23 +4561,23 @@ func (_c *SDK_Journal_Call) RunAndReturn(run func(string, string, string, pkgsdk
 	return _c
 }
 
-// ListChannelUsers provides a mock function with given fields: channelID, domainID, pm, token
-func (_m *SDK) ListChannelUsers(channelID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.UsersPage, errors.SDKError) {
+// ListChannelMembers provides a mock function with given fields: channelID, domainID, pm, token
+func (_m *SDK) ListChannelMembers(channelID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.EntityMembersPage, errors.SDKError) {
 	ret := _m.Called(channelID, domainID, pm, token)
 
 	if len(ret) == 0 {
-		panic("no return value specified for ListChannelUsers")
+		panic("no return value specified for ListChannelMembers")
 	}
 
-	var r0 pkgsdk.UsersPage
+	var r0 pkgsdk.EntityMembersPage
 	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)); ok {
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)); ok {
 		return rf(channelID, domainID, pm, token)
 	}
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.UsersPage); ok {
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.EntityMembersPage); ok {
 		r0 = rf(channelID, domainID, pm, token)
 	} else {
-		r0 = ret.Get(0).(pkgsdk.UsersPage)
+		r0 = ret.Get(0).(pkgsdk.EntityMembersPage)
 	}
 
 	if rf, ok := ret.Get(1).(func(string, string, pkgsdk.PageMetadata, string) errors.SDKError); ok {
@@ -4591,54 +4591,54 @@ func (_m *SDK) ListChannelUsers(channelID string, domainID string, pm pkgsdk.Pag
 	return r0, r1
 }
 
-// SDK_ListChannelUsers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListChannelUsers'
-type SDK_ListChannelUsers_Call struct {
+// SDK_ListChannelMembers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListChannelMembers'
+type SDK_ListChannelMembers_Call struct {
 	*mock.Call
 }
 
-// ListChannelUsers is a helper method to define mock.On call
+// ListChannelMembers is a helper method to define mock.On call
 //   - channelID string
 //   - domainID string
 //   - pm pkgsdk.PageMetadata
 //   - token string
-func (_e *SDK_Expecter) ListChannelUsers(channelID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListChannelUsers_Call {
-	return &SDK_ListChannelUsers_Call{Call: _e.mock.On("ListChannelUsers", channelID, domainID, pm, token)}
+func (_e *SDK_Expecter) ListChannelMembers(channelID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListChannelMembers_Call {
+	return &SDK_ListChannelMembers_Call{Call: _e.mock.On("ListChannelMembers", channelID, domainID, pm, token)}
 }
 
-func (_c *SDK_ListChannelUsers_Call) Run(run func(channelID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListChannelUsers_Call {
+func (_c *SDK_ListChannelMembers_Call) Run(run func(channelID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListChannelMembers_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		run(args[0].(string), args[1].(string), args[2].(pkgsdk.PageMetadata), args[3].(string))
 	})
 	return _c
 }
 
-func (_c *SDK_ListChannelUsers_Call) Return(_a0 pkgsdk.UsersPage, _a1 errors.SDKError) *SDK_ListChannelUsers_Call {
+func (_c *SDK_ListChannelMembers_Call) Return(_a0 pkgsdk.EntityMembersPage, _a1 errors.SDKError) *SDK_ListChannelMembers_Call {
 	_c.Call.Return(_a0, _a1)
 	return _c
 }
 
-func (_c *SDK_ListChannelUsers_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)) *SDK_ListChannelUsers_Call {
+func (_c *SDK_ListChannelMembers_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)) *SDK_ListChannelMembers_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// ListClientUsers provides a mock function with given fields: clientID, domainID, pm, token
-func (_m *SDK) ListClientUsers(clientID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.UsersPage, errors.SDKError) {
+// ListClientMembers provides a mock function with given fields: clientID, domainID, pm, token
+func (_m *SDK) ListClientMembers(clientID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.EntityMembersPage, errors.SDKError) {
 	ret := _m.Called(clientID, domainID, pm, token)
 
 	if len(ret) == 0 {
-		panic("no return value specified for ListClientUsers")
+		panic("no return value specified for ListClientMembers")
 	}
 
-	var r0 pkgsdk.UsersPage
+	var r0 pkgsdk.EntityMembersPage
 	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)); ok {
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)); ok {
 		return rf(clientID, domainID, pm, token)
 	}
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.UsersPage); ok {
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.EntityMembersPage); ok {
 		r0 = rf(clientID, domainID, pm, token)
 	} else {
-		r0 = ret.Get(0).(pkgsdk.UsersPage)
+		r0 = ret.Get(0).(pkgsdk.EntityMembersPage)
 	}
 
 	if rf, ok := ret.Get(1).(func(string, string, pkgsdk.PageMetadata, string) errors.SDKError); ok {
@@ -4652,54 +4652,54 @@ func (_m *SDK) ListClientUsers(clientID string, domainID string, pm pkgsdk.PageM
 	return r0, r1
 }
 
-// SDK_ListClientUsers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListClientUsers'
-type SDK_ListClientUsers_Call struct {
+// SDK_ListClientMembers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListClientMembers'
+type SDK_ListClientMembers_Call struct {
 	*mock.Call
 }
 
-// ListClientUsers is a helper method to define mock.On call
+// ListClientMembers is a helper method to define mock.On call
 //   - clientID string
 //   - domainID string
 //   - pm pkgsdk.PageMetadata
 //   - token string
-func (_e *SDK_Expecter) ListClientUsers(clientID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListClientUsers_Call {
-	return &SDK_ListClientUsers_Call{Call: _e.mock.On("ListClientUsers", clientID, domainID, pm, token)}
+func (_e *SDK_Expecter) ListClientMembers(clientID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListClientMembers_Call {
+	return &SDK_ListClientMembers_Call{Call: _e.mock.On("ListClientMembers", clientID, domainID, pm, token)}
 }
 
-func (_c *SDK_ListClientUsers_Call) Run(run func(clientID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListClientUsers_Call {
+func (_c *SDK_ListClientMembers_Call) Run(run func(clientID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListClientMembers_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		run(args[0].(string), args[1].(string), args[2].(pkgsdk.PageMetadata), args[3].(string))
 	})
 	return _c
 }
 
-func (_c *SDK_ListClientUsers_Call) Return(_a0 pkgsdk.UsersPage, _a1 errors.SDKError) *SDK_ListClientUsers_Call {
+func (_c *SDK_ListClientMembers_Call) Return(_a0 pkgsdk.EntityMembersPage, _a1 errors.SDKError) *SDK_ListClientMembers_Call {
 	_c.Call.Return(_a0, _a1)
 	return _c
 }
 
-func (_c *SDK_ListClientUsers_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)) *SDK_ListClientUsers_Call {
+func (_c *SDK_ListClientMembers_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)) *SDK_ListClientMembers_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// ListDomainUsers provides a mock function with given fields: domainID, pm, token
-func (_m *SDK) ListDomainUsers(domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.UsersPage, errors.SDKError) {
+// ListDomainMembers provides a mock function with given fields: domainID, pm, token
+func (_m *SDK) ListDomainMembers(domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.EntityMembersPage, errors.SDKError) {
 	ret := _m.Called(domainID, pm, token)
 
 	if len(ret) == 0 {
-		panic("no return value specified for ListDomainUsers")
+		panic("no return value specified for ListDomainMembers")
 	}
 
-	var r0 pkgsdk.UsersPage
+	var r0 pkgsdk.EntityMembersPage
 	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)); ok {
+	if rf, ok := ret.Get(0).(func(string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)); ok {
 		return rf(domainID, pm, token)
 	}
-	if rf, ok := ret.Get(0).(func(string, pkgsdk.PageMetadata, string) pkgsdk.UsersPage); ok {
+	if rf, ok := ret.Get(0).(func(string, pkgsdk.PageMetadata, string) pkgsdk.EntityMembersPage); ok {
 		r0 = rf(domainID, pm, token)
 	} else {
-		r0 = ret.Get(0).(pkgsdk.UsersPage)
+		r0 = ret.Get(0).(pkgsdk.EntityMembersPage)
 	}
 
 	if rf, ok := ret.Get(1).(func(string, pkgsdk.PageMetadata, string) errors.SDKError); ok {
@@ -4713,116 +4713,57 @@ func (_m *SDK) ListDomainUsers(domainID string, pm pkgsdk.PageMetadata, token st
 	return r0, r1
 }
 
-// SDK_ListDomainUsers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListDomainUsers'
-type SDK_ListDomainUsers_Call struct {
+// SDK_ListDomainMembers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListDomainMembers'
+type SDK_ListDomainMembers_Call struct {
 	*mock.Call
 }
 
-// ListDomainUsers is a helper method to define mock.On call
+// ListDomainMembers is a helper method to define mock.On call
 //   - domainID string
 //   - pm pkgsdk.PageMetadata
 //   - token string
-func (_e *SDK_Expecter) ListDomainUsers(domainID interface{}, pm interface{}, token interface{}) *SDK_ListDomainUsers_Call {
-	return &SDK_ListDomainUsers_Call{Call: _e.mock.On("ListDomainUsers", domainID, pm, token)}
+func (_e *SDK_Expecter) ListDomainMembers(domainID interface{}, pm interface{}, token interface{}) *SDK_ListDomainMembers_Call {
+	return &SDK_ListDomainMembers_Call{Call: _e.mock.On("ListDomainMembers", domainID, pm, token)}
 }
 
-func (_c *SDK_ListDomainUsers_Call) Run(run func(domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListDomainUsers_Call {
+func (_c *SDK_ListDomainMembers_Call) Run(run func(domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListDomainMembers_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		run(args[0].(string), args[1].(pkgsdk.PageMetadata), args[2].(string))
 	})
 	return _c
 }
 
-func (_c *SDK_ListDomainUsers_Call) Return(_a0 pkgsdk.UsersPage, _a1 errors.SDKError) *SDK_ListDomainUsers_Call {
-	_c.Call.Return(_a0, _a1)
-	return _c
-}
-
-func (_c *SDK_ListDomainUsers_Call) RunAndReturn(run func(string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)) *SDK_ListDomainUsers_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// ListSubscriptions provides a mock function with given fields: pm, token
-func (_m *SDK) ListSubscriptions(pm sdk.PageMetadata, token string) (sdk.SubscriptionPage, errors.SDKError) {
-	ret := _m.Called(pm, token)
-
-	if len(ret) == 0 {
-		panic("no return value specified for ListSubscriptions")
-	}
-
-	var r0 sdk.SubscriptionPage
-	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(sdk.PageMetadata, string) (sdk.SubscriptionPage, errors.SDKError)); ok {
-		return rf(pm, token)
-	}
-	if rf, ok := ret.Get(0).(func(sdk.PageMetadata, string) sdk.SubscriptionPage); ok {
-		r0 = rf(pm, token)
-	} else {
-		r0 = ret.Get(0).(sdk.SubscriptionPage)
-	}
-
-	if rf, ok := ret.Get(1).(func(sdk.PageMetadata, string) errors.SDKError); ok {
-		r1 = rf(pm, token)
-	} else {
-		if ret.Get(1) != nil {
-			r1 = ret.Get(1).(errors.SDKError)
-		}
-	}
-
-	return r0, r1
-}
-
-// SDK_ListSubscriptions_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListSubscriptions'
-type SDK_ListSubscriptions_Call struct {
-	*mock.Call
-}
-
-// ListSubscriptions is a helper method to define mock.On call
-//   - pm sdk.PageMetadata
-//   - token string
-func (_e *SDK_Expecter) ListSubscriptions(pm interface{}, token interface{}) *SDK_ListSubscriptions_Call {
-	return &SDK_ListSubscriptions_Call{Call: _e.mock.On("ListSubscriptions", pm, token)}
-}
-
-func (_c *SDK_ListSubscriptions_Call) Run(run func(pm sdk.PageMetadata, token string)) *SDK_ListSubscriptions_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(sdk.PageMetadata), args[1].(string))
-	})
-	return _c
-}
-
-func (_c *SDK_ListSubscriptions_Call) Return(_a0 sdk.SubscriptionPage, _a1 errors.SDKError) *SDK_ListSubscriptions_Call {
+func (_c *SDK_ListDomainMembers_Call) Return(_a0 pkgsdk.EntityMembersPage, _a1 errors.SDKError) *SDK_ListDomainMembers_Call {
 	_c.Call.Return(_a0, _a1)
 	return _c
 }
 
-func (_c *SDK_ListSubscriptions_Call) RunAndReturn(run func(sdk.PageMetadata, string) (sdk.SubscriptionPage, errors.SDKError)) *SDK_ListSubscriptions_Call {
+func (_c *SDK_ListDomainMembers_Call) RunAndReturn(run func(string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)) *SDK_ListDomainMembers_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// ListUserClients provides a mock function with given fields: userID, domainID, pm, token
-func (_m *SDK) ListUserClients(userID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.ClientsPage, errors.SDKError) {
-	ret := _m.Called(userID, domainID, pm, token)
+// ListGroupMembers provides a mock function with given fields: groupID, domainID, pm, token
+func (_m *SDK) ListGroupMembers(groupID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.EntityMembersPage, errors.SDKError) {
+	ret := _m.Called(groupID, domainID, pm, token)
 
 	if len(ret) == 0 {
-		panic("no return value specified for ListUserClients")
+		panic("no return value specified for ListGroupMembers")
 	}
 
-	var r0 pkgsdk.ClientsPage
+	var r0 pkgsdk.EntityMembersPage
 	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.ClientsPage, errors.SDKError)); ok {
-		return rf(userID, domainID, pm, token)
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)); ok {
+		return rf(groupID, domainID, pm, token)
 	}
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.ClientsPage); ok {
-		r0 = rf(userID, domainID, pm, token)
+	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.EntityMembersPage); ok {
+		r0 = rf(groupID, domainID, pm, token)
 	} else {
-		r0 = ret.Get(0).(pkgsdk.ClientsPage)
+		r0 = ret.Get(0).(pkgsdk.EntityMembersPage)
 	}
 
 	if rf, ok := ret.Get(1).(func(string, string, pkgsdk.PageMetadata, string) errors.SDKError); ok {
-		r1 = rf(userID, domainID, pm, token)
+		r1 = rf(groupID, domainID, pm, token)
 	} else {
 		if ret.Get(1) != nil {
 			r1 = ret.Get(1).(errors.SDKError)
@@ -4832,58 +4773,58 @@ func (_m *SDK) ListUserClients(userID string, domainID string, pm pkgsdk.PageMet
 	return r0, r1
 }
 
-// SDK_ListUserClients_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListUserClients'
-type SDK_ListUserClients_Call struct {
+// SDK_ListGroupMembers_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListGroupMembers'
+type SDK_ListGroupMembers_Call struct {
 	*mock.Call
 }
 
-// ListUserClients is a helper method to define mock.On call
-//   - userID string
+// ListGroupMembers is a helper method to define mock.On call
+//   - groupID string
 //   - domainID string
 //   - pm pkgsdk.PageMetadata
 //   - token string
-func (_e *SDK_Expecter) ListUserClients(userID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListUserClients_Call {
-	return &SDK_ListUserClients_Call{Call: _e.mock.On("ListUserClients", userID, domainID, pm, token)}
+func (_e *SDK_Expecter) ListGroupMembers(groupID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_ListGroupMembers_Call {
+	return &SDK_ListGroupMembers_Call{Call: _e.mock.On("ListGroupMembers", groupID, domainID, pm, token)}
 }
 
-func (_c *SDK_ListUserClients_Call) Run(run func(userID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListUserClients_Call {
+func (_c *SDK_ListGroupMembers_Call) Run(run func(groupID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_ListGroupMembers_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		run(args[0].(string), args[1].(string), args[2].(pkgsdk.PageMetadata), args[3].(string))
 	})
 	return _c
 }
 
-func (_c *SDK_ListUserClients_Call) Return(_a0 pkgsdk.ClientsPage, _a1 errors.SDKError) *SDK_ListUserClients_Call {
+func (_c *SDK_ListGroupMembers_Call) Return(_a0 pkgsdk.EntityMembersPage, _a1 errors.SDKError) *SDK_ListGroupMembers_Call {
 	_c.Call.Return(_a0, _a1)
 	return _c
 }
 
-func (_c *SDK_ListUserClients_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.ClientsPage, errors.SDKError)) *SDK_ListUserClients_Call {
+func (_c *SDK_ListGroupMembers_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.EntityMembersPage, errors.SDKError)) *SDK_ListGroupMembers_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// Members provides a mock function with given fields: groupID, domainID, pm, token
-func (_m *SDK) Members(groupID string, domainID string, pm pkgsdk.PageMetadata, token string) (pkgsdk.UsersPage, errors.SDKError) {
-	ret := _m.Called(groupID, domainID, pm, token)
+// ListSubscriptions provides a mock function with given fields: pm, token
+func (_m *SDK) ListSubscriptions(pm sdk.PageMetadata, token string) (sdk.SubscriptionPage, errors.SDKError) {
+	ret := _m.Called(pm, token)
 
 	if len(ret) == 0 {
-		panic("no return value specified for Members")
+		panic("no return value specified for ListSubscriptions")
 	}
 
-	var r0 pkgsdk.UsersPage
+	var r0 sdk.SubscriptionPage
 	var r1 errors.SDKError
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)); ok {
-		return rf(groupID, domainID, pm, token)
+	if rf, ok := ret.Get(0).(func(sdk.PageMetadata, string) (sdk.SubscriptionPage, errors.SDKError)); ok {
+		return rf(pm, token)
 	}
-	if rf, ok := ret.Get(0).(func(string, string, pkgsdk.PageMetadata, string) pkgsdk.UsersPage); ok {
-		r0 = rf(groupID, domainID, pm, token)
+	if rf, ok := ret.Get(0).(func(sdk.PageMetadata, string) sdk.SubscriptionPage); ok {
+		r0 = rf(pm, token)
 	} else {
-		r0 = ret.Get(0).(pkgsdk.UsersPage)
+		r0 = ret.Get(0).(sdk.SubscriptionPage)
 	}
 
-	if rf, ok := ret.Get(1).(func(string, string, pkgsdk.PageMetadata, string) errors.SDKError); ok {
-		r1 = rf(groupID, domainID, pm, token)
+	if rf, ok := ret.Get(1).(func(sdk.PageMetadata, string) errors.SDKError); ok {
+		r1 = rf(pm, token)
 	} else {
 		if ret.Get(1) != nil {
 			r1 = ret.Get(1).(errors.SDKError)
@@ -4893,33 +4834,31 @@ func (_m *SDK) Members(groupID string, domainID string, pm pkgsdk.PageMetadata,
 	return r0, r1
 }
 
-// SDK_Members_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Members'
-type SDK_Members_Call struct {
+// SDK_ListSubscriptions_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListSubscriptions'
+type SDK_ListSubscriptions_Call struct {
 	*mock.Call
 }
 
-// Members is a helper method to define mock.On call
-//   - groupID string
-//   - domainID string
-//   - pm pkgsdk.PageMetadata
+// ListSubscriptions is a helper method to define mock.On call
+//   - pm sdk.PageMetadata
 //   - token string
-func (_e *SDK_Expecter) Members(groupID interface{}, domainID interface{}, pm interface{}, token interface{}) *SDK_Members_Call {
-	return &SDK_Members_Call{Call: _e.mock.On("Members", groupID, domainID, pm, token)}
+func (_e *SDK_Expecter) ListSubscriptions(pm interface{}, token interface{}) *SDK_ListSubscriptions_Call {
+	return &SDK_ListSubscriptions_Call{Call: _e.mock.On("ListSubscriptions", pm, token)}
 }
 
-func (_c *SDK_Members_Call) Run(run func(groupID string, domainID string, pm pkgsdk.PageMetadata, token string)) *SDK_Members_Call {
+func (_c *SDK_ListSubscriptions_Call) Run(run func(pm sdk.PageMetadata, token string)) *SDK_ListSubscriptions_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(string), args[1].(string), args[2].(pkgsdk.PageMetadata), args[3].(string))
+		run(args[0].(sdk.PageMetadata), args[1].(string))
 	})
 	return _c
 }
 
-func (_c *SDK_Members_Call) Return(_a0 pkgsdk.UsersPage, _a1 errors.SDKError) *SDK_Members_Call {
+func (_c *SDK_ListSubscriptions_Call) Return(_a0 sdk.SubscriptionPage, _a1 errors.SDKError) *SDK_ListSubscriptions_Call {
 	_c.Call.Return(_a0, _a1)
 	return _c
 }
 
-func (_c *SDK_Members_Call) RunAndReturn(run func(string, string, pkgsdk.PageMetadata, string) (pkgsdk.UsersPage, errors.SDKError)) *SDK_Members_Call {
+func (_c *SDK_ListSubscriptions_Call) RunAndReturn(run func(sdk.PageMetadata, string) (sdk.SubscriptionPage, errors.SDKError)) *SDK_ListSubscriptions_Call {
 	_c.Call.Return(run)
 	return _c
 }
diff --git a/pkg/sdk/sdk.go b/pkg/sdk/sdk.go
index d182f1675..3fab4cfa0 100644
--- a/pkg/sdk/sdk.go
+++ b/pkg/sdk/sdk.go
@@ -213,7 +213,6 @@ type Config struct {
 	GroupsURL      string
 	ChannelsURL    string
 	DomainsURL     string
-	InvitationsURL string
 	JournalURL     string
 	HostURL        string
 
@@ -232,7 +231,6 @@ func NewSDK(conf Config) SDK {
 		GroupsURL:      conf.GroupsURL,
 		ChannelsURL:    conf.ChannelsURL,
 		DomainsURL:     conf.DomainsURL,
-		InvitationsURL: conf.InvitationsURL,
 		JournalURL:     conf.JournalURL,
 		HostURL:        conf.HostURL,
 
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/provision/config_test.go b/provision/config_test.go
index 7ba27c6dc..cd9c47589 100644
--- a/provision/config_test.go
+++ b/provision/config_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/absmach/magistrala/provision"
 	"github.com/absmach/supermq/channels"
 	"github.com/absmach/supermq/clients"
+	"github.com/absmach/supermq/pkg/connections"
 	"github.com/absmach/supermq/pkg/errors"
 	"github.com/pelletier/go-toml"
 	"github.com/stretchr/testify/assert"
@@ -39,7 +40,9 @@ var (
 				Metadata: map[string]interface{}{
 					"test": "test",
 				},
-				Permissions: []string{"test"},
+				Actions:                   []string{},
+				AccessProviderRoleActions: []string{},
+				ConnectionTypes:           []connections.ConnType{},
 			},
 		},
 		Channels: []channels.Channel{
@@ -50,7 +53,9 @@ var (
 				Metadata: map[string]interface{}{
 					"test": "test",
 				},
-				Permissions: []string{"test"},
+				Actions:                   []string{},
+				AccessProviderRoleActions: []string{},
+				ConnectionTypes:           []connections.ConnType{},
 			},
 		},
 		Cert:          provision.Cert{},
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