diff --git a/Makefile b/Makefile
index 3c2095cf4..536c96aaf 100644
--- a/Makefile
+++ b/Makefile
@@ -197,7 +197,14 @@ FILTERED_SERVICES = $(filter-out $(RUN_ADDON_ARGS), $(SERVICES))
 
 all: $(SERVICES)
 
-.PHONY: all $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs check_fluxmq_service_certs check_re_trace_key test_api mocks
+.PHONY: all help $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs check_fluxmq_service_certs check_re_trace_key test_api mocks
+
+help:
+	@printf 'Usage:\n  make <target> [VARIABLE=value ...]\n\nAvailable targets:\n'
+	@$(MAKE) -qpRr : 2>/dev/null | \
+		awk -F: '/^[[:alnum:]_][^$$#\/\t=]*:([^=]|$$)/ { split($$1, targets, /[[:space:]]+/); for (i in targets) if (targets[i] != "") print targets[i] }' | \
+		LC_ALL=C sort -u | \
+		awk '$$0 != "Makefile" { printf "  make %s\n", $$0 }'
 
 clean:
 	rm -rf ${BUILD_DIR}
@@ -349,16 +356,28 @@ else
 	$(eval GRPC_MTLS :=)
 endif
 
-# The Rules Engine reaches FluxMQ on the mTLS local listener, and compose mounts
-# both sides of that connection. The certificates and the principal secret are
+# Internal services reach FluxMQ on the mTLS local listener, and Compose mounts
+# both sides of those connections. Certificates and principal secrets are
 # generated rather than committed, so make them before anything binds them.
 check_fluxmq_service_certs:
 ifeq ("$(wildcard docker/ssl/certs/re-fluxmq-client.crt)","")
 	$(MAKE) -C docker/ssl fluxmq_service_certs
 endif
+ifeq ("$(wildcard docker/ssl/certs/timescale-writer-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl timescale_writer_fluxmq_client_cert
+endif
+ifeq ("$(wildcard docker/ssl/certs/postgres-writer-fluxmq-client.crt)","")
+	$(MAKE) -C docker/ssl postgres_writer_fluxmq_client_cert
+endif
 ifeq ("$(wildcard docker/fluxmq/secrets/re-current)","")
 	$(MAKE) -C docker/ssl fluxmq_service_secret
 endif
+ifeq ("$(wildcard docker/fluxmq/secrets/timescale-writer-current)","")
+	$(MAKE) -C docker/ssl timescale_writer_fluxmq_service_secret
+endif
+ifeq ("$(wildcard docker/fluxmq/secrets/postgres-writer-current)","")
+	$(MAKE) -C docker/ssl postgres_writer_fluxmq_service_secret
+endif
 
 check_re_trace_key:
 ifeq ("$(wildcard docker/re/secrets/trace.key)","")
diff --git a/README.md b/README.md
index ff90544a5..2896a7743 100644
--- a/README.md
+++ b/README.md
@@ -300,34 +300,54 @@ make check_certs
 
 This creates whatever is missing and leaves anything already present alone:
 
-| Path                                               | What it is                                                                    |
-| -------------------------------------------------- | ----------------------------------------------------------------------------- |
-| `docker/ssl/certs/fluxmq-service-server.{crt,key}` | Server certificate for FluxMQ's mTLS service listener                         |
-| `docker/ssl/certs/re-fluxmq-client.{crt,key}`      | Client certificate whose URI SAN identifies the Rules Engine to that listener |
-| `docker/fluxmq/secrets/re-current`                 | The principal secret, derived from `MG_RE_BROKER_SECRET` in `docker/.env`     |
-| `docker/re/secrets/trace.key`                      | HMAC key the Rules Engine signs its loop-detection traces with                |
+| Path                                                        | What it is                                                                |
+| ----------------------------------------------------------- | ------------------------------------------------------------------------- |
+| `docker/ssl/certs/fluxmq-service-server.{crt,key}`          | Server certificate for FluxMQ's mTLS service listener                     |
+| `docker/ssl/certs/re-fluxmq-client.{crt,key}`               | Client certificate whose URI SAN identifies the Rules Engine              |
+| `docker/ssl/certs/timescale-writer-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Timescale writer          |
+| `docker/ssl/certs/postgres-writer-fluxmq-client.{crt,key}`  | Client certificate whose URI SAN identifies the Postgres writer           |
+| `docker/fluxmq/secrets/re-current`                          | Rules Engine principal secret, from `MG_RE_BROKER_SECRET`                 |
+| `docker/fluxmq/secrets/timescale-writer-current`            | Timescale writer secret, from `MG_TIMESCALE_WRITER_BROKER_SECRET`         |
+| `docker/fluxmq/secrets/postgres-writer-current`             | Postgres writer secret, from `MG_POSTGRES_WRITER_BROKER_SECRET`           |
+| `docker/re/secrets/trace.key`                               | HMAC key the Rules Engine signs its loop-detection traces with            |
 
-Both certificates are issued by the development CA committed at
+Internal services reach the broker as *local principals* rather than as ordinary
+clients: each presents a client certificate whose URI SAN names it, plus a SASL
+secret, and the broker grants it only what it needs — the Rules Engine consumes
+`m` and republishes under it, the writers only subscribe to `writers`. The
+principals are declared in `docker/fluxmq/node{1,2,3}.yaml`, and adding a service
+means adding an entry there alongside its certificate and secret.
+
+The certificates are issued by the development CA committed at
 `docker/ssl/certs/ca.crt`, so no extra setup is needed for a local run. The
 generated material is gitignored.
 
 The server certificate is issued for `fluxmq` and `fluxmq-node{1,2,3}`, which
-covers both this Compose stack and a single-node deployment. Point
-`MG_RE_BROKER_URL` at a host outside that set and the Rules Engine fails its
-TLS verification with `certificate is valid for ...`; add the name to
+covers both this Compose stack and a single-node deployment. Point any
+`MG_*_BROKER_URL` at a host outside that set and the service fails its TLS
+verification with `certificate is valid for ...`; add the name to
 `FLUXMQ_SERVICE_SERVER_CERT_CONFIG` in `docker/ssl/Makefile` and reissue:
 
 ```bash
-rm -f docker/ssl/certs/fluxmq-service-server.* docker/ssl/certs/re-fluxmq-client.*
+rm -f docker/ssl/certs/fluxmq-service-server.* \
+  docker/ssl/certs/re-fluxmq-client.* \
+  docker/ssl/certs/timescale-writer-fluxmq-client.* \
+  docker/ssl/certs/postgres-writer-fluxmq-client.*
 make -C docker/ssl fluxmq_service_certs
 ```
 
-`make check_certs` skips this pair when it already exists, so a stale
-certificate has to be removed rather than merely re-run.
+`make check_certs` skips certificates that already exist, so stale certificates
+have to be removed rather than merely re-running the target.
 
-Two of these are stateful, not merely derived. `re-current` must stay equal to
-`MG_RE_BROKER_SECRET`: change the variable and re-run `make -C docker/ssl
-fluxmq_service_secret`, or the Rules Engine fails to authenticate.
+Each local-principal secret must stay equal to the corresponding value in
+`docker/.env`; a mismatch fails that service's broker authentication. After
+changing one, re-run its target:
+
+| Variable                            | Target                                    |
+| ----------------------------------- | ----------------------------------------- |
+| `MG_RE_BROKER_SECRET`               | `fluxmq_service_secret`                   |
+| `MG_TIMESCALE_WRITER_BROKER_SECRET` | `timescale_writer_fluxmq_service_secret`  |
+| `MG_POSTGRES_WRITER_BROKER_SECRET`  | `postgres_writer_fluxmq_service_secret`   |
 `trace.key` is created once and preserved on later runs — replacing it while
 messages are in flight would invalidate the rule traces they already carry, so
 delete it only deliberately. Every Rules Engine replica must read the same key.
diff --git a/cmd/postgres-writer/main.go b/cmd/postgres-writer/main.go
index e8ece3e30..a5caf595d 100644
--- a/cmd/postgres-writer/main.go
+++ b/cmd/postgres-writer/main.go
@@ -20,8 +20,9 @@ import (
 	writerpg "github.com/absmach/magistrala/consumers/writers/postgres"
 	mglog "github.com/absmach/magistrala/logger"
 	jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
+	"github.com/absmach/magistrala/pkg/messaging"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
-	brokers "github.com/absmach/magistrala/pkg/messaging/writers"
+	"github.com/absmach/magistrala/pkg/messaging/writers"
 	pgclient "github.com/absmach/magistrala/pkg/postgres"
 	"github.com/absmach/magistrala/pkg/prometheus"
 	"github.com/absmach/magistrala/pkg/server"
@@ -41,13 +42,16 @@ const (
 )
 
 type config struct {
-	LogLevel      string  `env:"MG_POSTGRES_WRITER_LOG_LEVEL"     envDefault:"info"`
-	ConfigPath    string  `env:"MG_POSTGRES_WRITER_CONFIG_PATH"   envDefault:"/config.toml"`
-	BrokerURL     string  `env:"MG_MESSAGE_BROKER_URL"            envDefault:"nats://localhost:4222"`
-	JaegerURL     url.URL `env:"MG_JAEGER_URL"                    envDefault:"http://localhost:4318/v1/traces"`
-	SendTelemetry bool    `env:"MG_SEND_TELEMETRY"                envDefault:"true"`
-	InstanceID    string  `env:"MG_POSTGRES_WRITER_INSTANCE_ID"   envDefault:""`
-	TraceRatio    float64 `env:"MG_JAEGER_TRACE_RATIO"            envDefault:"1.0"`
+	LogLevel         string  `env:"MG_POSTGRES_WRITER_LOG_LEVEL"          envDefault:"info"`
+	ConfigPath       string  `env:"MG_POSTGRES_WRITER_CONFIG_PATH"        envDefault:"/config.toml"`
+	BrokerURL        string  `env:"MG_MESSAGE_BROKER_URL"                 envDefault:"nats://localhost:4222"`
+	BrokerClientCert string  `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_CERT" envDefault:""`
+	BrokerClientKey  string  `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_KEY"  envDefault:""`
+	BrokerCACerts    string  `env:"MG_POSTGRES_WRITER_BROKER_CA_CERTS"    envDefault:""`
+	JaegerURL        url.URL `env:"MG_JAEGER_URL"                         envDefault:"http://localhost:4318/v1/traces"`
+	SendTelemetry    bool    `env:"MG_SEND_TELEMETRY"                     envDefault:"true"`
+	InstanceID       string  `env:"MG_POSTGRES_WRITER_INSTANCE_ID"        envDefault:""`
+	TraceRatio       float64 `env:"MG_JAEGER_TRACE_RATIO"                 envDefault:"1.0"`
 }
 
 func main() {
@@ -107,7 +111,16 @@ func main() {
 	}()
 	tracer := tp.Tracer(svcName)
 
-	pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+	// The mTLS client identity is optional: brokers that expose an unauthenticated
+	// listener need none. Passing the option with empty paths would fail the
+	// connection outright, so only ask for it once one of them is configured. A
+	// partial configuration still fails, since half an identity is not a usable one.
+	var brokerOpts []messaging.Option
+	if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
+		brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
+	}
+
+	pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
 		exitCode = 1
@@ -119,7 +132,7 @@ func main() {
 	repo := newService(db, logger)
 	repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
 
-	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
+	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
 		logger.Error(fmt.Sprintf("failed to create Postgres writer: %s", err))
 		exitCode = 1
 		return
diff --git a/cmd/timescale-writer/main.go b/cmd/timescale-writer/main.go
index b675ddaec..578aa1ca9 100644
--- a/cmd/timescale-writer/main.go
+++ b/cmd/timescale-writer/main.go
@@ -20,8 +20,9 @@ import (
 	"github.com/absmach/magistrala/consumers/writers/timescale"
 	mglog "github.com/absmach/magistrala/logger"
 	jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
+	"github.com/absmach/magistrala/pkg/messaging"
 	brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
-	brokers "github.com/absmach/magistrala/pkg/messaging/writers"
+	"github.com/absmach/magistrala/pkg/messaging/writers"
 	pgclient "github.com/absmach/magistrala/pkg/postgres"
 	"github.com/absmach/magistrala/pkg/prometheus"
 	"github.com/absmach/magistrala/pkg/server"
@@ -41,13 +42,16 @@ const (
 )
 
 type config struct {
-	LogLevel      string  `env:"MG_TIMESCALE_WRITER_LOG_LEVEL"     envDefault:"info"`
-	ConfigPath    string  `env:"MG_TIMESCALE_WRITER_CONFIG_PATH"   envDefault:"/config.toml"`
-	BrokerURL     string  `env:"MG_MESSAGE_BROKER_URL"            envDefault:"nats://localhost:4222"`
-	JaegerURL     url.URL `env:"MG_JAEGER_URL"                    envDefault:"http://localhost:4318/v1/traces"`
-	SendTelemetry bool    `env:"MG_SEND_TELEMETRY"                envDefault:"true"`
-	InstanceID    string  `env:"MG_TIMESCALE_WRITER_INSTANCE_ID"   envDefault:""`
-	TraceRatio    float64 `env:"MG_JAEGER_TRACE_RATIO"            envDefault:"1.0"`
+	LogLevel         string  `env:"MG_TIMESCALE_WRITER_LOG_LEVEL"          envDefault:"info"`
+	ConfigPath       string  `env:"MG_TIMESCALE_WRITER_CONFIG_PATH"        envDefault:"/config.toml"`
+	BrokerURL        string  `env:"MG_MESSAGE_BROKER_URL"                  envDefault:"nats://localhost:4222"`
+	BrokerClientCert string  `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT" envDefault:""`
+	BrokerClientKey  string  `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY"  envDefault:""`
+	BrokerCACerts    string  `env:"MG_TIMESCALE_WRITER_BROKER_CA_CERTS"    envDefault:""`
+	JaegerURL        url.URL `env:"MG_JAEGER_URL"                          envDefault:"http://localhost:4318/v1/traces"`
+	SendTelemetry    bool    `env:"MG_SEND_TELEMETRY"                      envDefault:"true"`
+	InstanceID       string  `env:"MG_TIMESCALE_WRITER_INSTANCE_ID"        envDefault:""`
+	TraceRatio       float64 `env:"MG_JAEGER_TRACE_RATIO"                  envDefault:"1.0"`
 }
 
 func main() {
@@ -112,7 +116,16 @@ func main() {
 	repo := newService(db, logger)
 	repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
 
-	pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
+	// The mTLS client identity is optional: brokers that expose an unauthenticated
+	// listener need none. Passing the option with empty paths would fail the
+	// connection outright, so only ask for it once one of them is configured. A
+	// partial configuration still fails, since half an identity is not a usable one.
+	var brokerOpts []messaging.Option
+	if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
+		brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
+	}
+
+	pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
 		exitCode = 1
@@ -121,7 +134,7 @@ func main() {
 	defer pubSub.Close()
 	pubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, pubSub)
 
-	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
+	if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
 		logger.Error(fmt.Sprintf("failed to create Timescale writer: %s", err))
 		exitCode = 1
 		return
diff --git a/docker/.env b/docker/.env
index 34f041c48..1eb94a617 100644
--- a/docker/.env
+++ b/docker/.env
@@ -502,6 +502,14 @@ MG_POSTGRES_WRITER_HTTP_PORT=9007
 MG_POSTGRES_WRITER_HTTP_SERVER_CERT=
 MG_POSTGRES_WRITER_HTTP_SERVER_KEY=
 MG_POSTGRES_WRITER_INSTANCE_ID=
+# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
+# certificate identifies the local principal and its SASL secret must match
+# docker/fluxmq/secrets/postgres-writer-current.
+MG_POSTGRES_WRITER_BROKER_SECRET=postgres-writer-fluxmq-local-principal-secret-change-me
+MG_POSTGRES_WRITER_BROKER_URL=amqps://postgres-writer:${MG_POSTGRES_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
+MG_POSTGRES_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/postgres-writer-fluxmq-client.crt
+MG_POSTGRES_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/postgres-writer-fluxmq-client.key
+MG_POSTGRES_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 
 ### Postgres Reader
 MG_POSTGRES_READER_LOG_LEVEL=debug
@@ -529,6 +537,14 @@ MG_TIMESCALE_WRITER_HTTP_PORT=9012
 MG_TIMESCALE_WRITER_HTTP_SERVER_CERT=
 MG_TIMESCALE_WRITER_HTTP_SERVER_KEY=
 MG_TIMESCALE_WRITER_INSTANCE_ID=
+# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
+# certificate identifies the local principal and its SASL secret must match
+# docker/fluxmq/secrets/timescale-writer-current.
+MG_TIMESCALE_WRITER_BROKER_SECRET=timescale-writer-fluxmq-local-principal-secret-change-me
+MG_TIMESCALE_WRITER_BROKER_URL=amqps://timescale-writer:${MG_TIMESCALE_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
+MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/timescale-writer-fluxmq-client.crt
+MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/timescale-writer-fluxmq-client.key
+MG_TIMESCALE_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
 
 ### Timescale Reader
 MG_TIMESCALE_READER_LOG_LEVEL=debug
diff --git a/docker/addons/postgres-writer/docker-compose.yaml b/docker/addons/postgres-writer/docker-compose.yaml
index 023867af1..38d1991c8 100644
--- a/docker/addons/postgres-writer/docker-compose.yaml
+++ b/docker/addons/postgres-writer/docker-compose.yaml
@@ -35,7 +35,10 @@ services:
     image: ghcr.io/absmach/magistrala/postgres-writer:${MG_RELEASE_TAG}
     container_name: magistrala-postgres-writer
     depends_on:
-      - postgres
+      postgres:
+        condition: service_started
+      fluxmq-node1:
+        condition: service_healthy
     restart: on-failure
     environment:
       MG_POSTGRES_WRITER_LOG_LEVEL: ${MG_POSTGRES_WRITER_LOG_LEVEL}
@@ -53,7 +56,10 @@ services:
       MG_POSTGRES_SSL_CERT: ${MG_POSTGRES_SSL_CERT}
       MG_POSTGRES_SSL_KEY: ${MG_POSTGRES_SSL_KEY}
       MG_POSTGRES_SSL_ROOT_CERT: ${MG_POSTGRES_SSL_ROOT_CERT}
-      MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
+      MG_MESSAGE_BROKER_URL: ${MG_POSTGRES_WRITER_BROKER_URL}
+      MG_POSTGRES_WRITER_BROKER_CLIENT_CERT: ${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}
+      MG_POSTGRES_WRITER_BROKER_CLIENT_KEY: ${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}
+      MG_POSTGRES_WRITER_BROKER_CA_CERTS: ${MG_POSTGRES_WRITER_BROKER_CA_CERTS}
       MG_JAEGER_URL: ${MG_JAEGER_URL}
       MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
       MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
@@ -63,4 +69,11 @@ services:
     networks:
       - magistrala-base-net
     volumes:
-      - ./config.toml:/config.toml
+      # Relative paths in an addon file resolve against the project directory,
+      # which is the base compose file's directory rather than this one.
+      - ./addons/postgres-writer/config.toml:${MG_POSTGRES_WRITER_CONFIG_PATH}
+      # The URI SAN in this client certificate identifies the Postgres writer
+      # to the FluxMQ local listener; the broker ACL only permits `writers`.
+      - ./ssl/certs/postgres-writer-fluxmq-client.crt:${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/postgres-writer-fluxmq-client.key:${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_POSTGRES_WRITER_BROKER_CA_CERTS}:ro
diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml
index 9d73f21bc..81f2a94e8 100644
--- a/docker/docker-compose.yaml
+++ b/docker/docker-compose.yaml
@@ -255,12 +255,14 @@ services:
         ipv4_address: 172.30.0.201
     volumes:
       - ./fluxmq/node1.yaml:/etc/fluxmq/config.yaml:ro
-      # Server certificate and CA for the mTLS service listener, plus the
-      # local-principal secret it authenticates the Rules Engine against.
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
       - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
       - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
       - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
       - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
       - magistrala-fluxmq-node1-volume:/tmp/fluxmq
 
   fluxmq-node2:
@@ -281,12 +283,14 @@ services:
         ipv4_address: 172.30.0.202
     volumes:
       - ./fluxmq/node2.yaml:/etc/fluxmq/config.yaml:ro
-      # Server certificate and CA for the mTLS service listener, plus the
-      # local-principal secret it authenticates the Rules Engine against.
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
       - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
       - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
       - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
       - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
       - magistrala-fluxmq-node2-volume:/tmp/fluxmq
 
   fluxmq-node3:
@@ -307,12 +311,14 @@ services:
         ipv4_address: 172.30.0.203
     volumes:
       - ./fluxmq/node3.yaml:/etc/fluxmq/config.yaml:ro
-      # Server certificate and CA for the mTLS service listener, plus the
-      # local-principal secret it authenticates the Rules Engine against.
+      # Server certificate and CA for the mTLS service listener, plus each
+      # local principal's SASL secret.
       - ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
       - ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
       - ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
       - ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
+      - ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
+      - ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
       - magistrala-fluxmq-node3-volume:/tmp/fluxmq
 
   fluxmq-auth:
@@ -557,8 +563,12 @@ services:
     networks:
       - magistrala-base-net
 
+  # Creates the image bucket with the broker's own image rather than pulling a
+  # second one. An aws-cli here cost a 130 MB image for a single call, and it
+  # was the only untagged reference in this file, so what it resolved to
+  # depended on the day it was pulled.
   seaweedfs-init:
-    image: amazon/aws-cli
+    image: chrislusf/seaweedfs:4.40
     container_name: magistrala-seaweedfs-init
     entrypoint: /bin/sh
     depends_on:
@@ -566,26 +576,46 @@ services:
     command:
       - -c
       - |
-        echo "[INIT] Waiting 20s for SeaweedFS S3 to be ready...";
-        sleep 20;
-        OUT=$(aws --endpoint-url http://seaweedfs-s3:8333 s3api create-bucket --bucket $${BUCKET} 2>&1);
-        EXIT=$$?;
-        if [ $$EXIT -eq 0 ]; then
-          echo "[INIT] Bucket $${BUCKET} created successfully.";
-        elif echo "$$OUT" | grep -q 'BucketAlreadyOwnedByYou\|BucketAlreadyExists'; then
-          echo "[INIT] Bucket $${BUCKET} already exists, skipping.";
-        else
-          echo "[INIT] Failed to create bucket $${BUCKET}: $$OUT" >&2;
-          exit 1;
-        fi
+        set -eu
+        # weed shell blocks indefinitely against an unreachable master, so the
+        # wait is a port check and every weed call is bounded; left to itself it
+        # would hang this container rather than fail it.
+        #
+        # 60 ticks is ~2 minutes while the store is merely slow to boot, and
+        # ~3 minutes to give up when it never appears, since a failing lookup
+        # costs about a second on top of the sleep.
+        tick=0;
+        while ! nc -z seaweedfs-s3 9333 2>/dev/null; do
+          tick=$$((tick + 1));
+          if [ $$tick -ge 60 ]; then
+            echo "[INIT] SeaweedFS master did not come up" >&2;
+            exit 1;
+          fi;
+          sleep 2;
+        done;
+        printf 's3.bucket.create -name %s\n' "$${BUCKET}" > /tmp/create.cmd;
+        printf 's3.bucket.list\n' > /tmp/list.cmd;
+        try=0;
+        while [ $$try -lt 3 ]; do
+          try=$$((try + 1));
+          timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/create.cmd > /tmp/create.out 2>&1 || true;
+          timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/list.cmd > /tmp/list.out 2>/dev/null || true;
+          # s3.bucket.create reports success even for an existing bucket, and
+          # weed shell exits 0 whatever the command did, so the listing is the
+          # only trustworthy signal that the bucket is really there.
+          if awk -v b="$${BUCKET}" '{sub(/^[ \t]+/,""); split($$0,f,/[ \t]/); if (f[1]==b) found=1} END{exit !found}' /tmp/list.out; then
+            echo "[INIT] Bucket $${BUCKET} present.";
+            exit 0;
+          fi;
+          sleep 3;
+        done;
+        echo "[INIT] Bucket $${BUCKET} was not created after $$try attempts" >&2;
+        cat /tmp/create.out >&2;
+        exit 1
     networks:
       - magistrala-base-net
     environment:
       BUCKET: ${MG_BACKEND_OBJECT_STORAGE_BUCKET}
-      AWS_ACCESS_KEY_ID: ${MG_BACKEND_OBJECT_STORAGE_ACCESS_KEY}
-      AWS_SECRET_ACCESS_KEY: ${MG_BACKEND_OBJECT_STORAGE_SECRET_KEY}
-      AWS_DEFAULT_REGION: ${MG_BACKEND_OBJECT_STORAGE_REGION}
-      AWS_EC2_METADATA_DISABLED: "true"
 
   timescale:
     image: timescale/timescaledb:2.29.0-pg16-oss
@@ -601,13 +631,18 @@ services:
       - magistrala-base-net
     volumes:
       - magistrala-timescale-writer-volume:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
+      interval: 5s
+      timeout: 3s
+      retries: 60
 
   timescale-reader:
     image: ghcr.io/absmach/magistrala/timescale-reader:${MG_RELEASE_TAG}
     container_name: magistrala-timescale-reader
     depends_on:
       timescale:
-        condition: service_started
+        condition: service_healthy
       atom-bootstrap:
         condition: service_completed_successfully
     restart: on-failure
@@ -686,7 +721,10 @@ services:
     image: ghcr.io/absmach/magistrala/timescale-writer:${MG_RELEASE_TAG}
     container_name: magistrala-timescale-writer
     depends_on:
-      - timescale
+      timescale:
+        condition: service_healthy
+      fluxmq-node1:
+        condition: service_healthy
     restart: on-failure
     environment:
       MG_TIMESCALE_WRITER_LOG_LEVEL: ${MG_TIMESCALE_WRITER_LOG_LEVEL}
@@ -704,7 +742,10 @@ services:
       MG_TIMESCALE_SSL_CERT: ${MG_TIMESCALE_SSL_CERT}
       MG_TIMESCALE_SSL_KEY: ${MG_TIMESCALE_SSL_KEY}
       MG_TIMESCALE_SSL_ROOT_CERT: ${MG_TIMESCALE_SSL_ROOT_CERT}
-      MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
+      MG_MESSAGE_BROKER_URL: ${MG_TIMESCALE_WRITER_BROKER_URL}
+      MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}
+      MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}
+      MG_TIMESCALE_WRITER_BROKER_CA_CERTS: ${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}
       MG_JAEGER_URL: ${MG_JAEGER_URL}
       MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
       MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
@@ -715,6 +756,11 @@ services:
       - magistrala-base-net
     volumes:
       - ./addons/timescale-writer/config.toml:${MG_TIMESCALE_WRITER_CONFIG_PATH}
+      # The URI SAN in this client certificate identifies the Timescale writer
+      # to the FluxMQ local listener; the broker ACL only permits `writers`.
+      - ./ssl/certs/timescale-writer-fluxmq-client.crt:${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}:ro
+      - ./ssl/certs/timescale-writer-fluxmq-client.key:${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}:ro
+      - ./ssl/certs/ca.crt:${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}:ro
   re-db:
     image: docker.io/postgres:18.4-alpine3.24
     container_name: magistrala-re-db
diff --git a/docker/fluxmq/node1.yaml b/docker/fluxmq/node1.yaml
index 5a4dd391b..17446fb43 100644
--- a/docker/fluxmq/node1.yaml
+++ b/docker/fluxmq/node1.yaml
@@ -143,7 +143,7 @@ queues:
 auth:
   # Principals admitted on the service listener. The certificate_uri_san must
   # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
-  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  # each secret file must hold the value configured by its service.
   local_principals:
     - name: "rules-engine"
       certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
@@ -164,6 +164,24 @@ auth:
         # The message stream the Rules Engine consumes.
         subscribe:
           - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"
diff --git a/docker/fluxmq/node2.yaml b/docker/fluxmq/node2.yaml
index 6dbe4fe57..a5274fb03 100644
--- a/docker/fluxmq/node2.yaml
+++ b/docker/fluxmq/node2.yaml
@@ -140,7 +140,7 @@ queues:
 auth:
   # Principals admitted on the service listener. The certificate_uri_san must
   # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
-  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  # each secret file must hold the value configured by its service.
   local_principals:
     - name: "rules-engine"
       certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
@@ -161,6 +161,24 @@ auth:
         # The message stream the Rules Engine consumes.
         subscribe:
           - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"
diff --git a/docker/fluxmq/node3.yaml b/docker/fluxmq/node3.yaml
index 20e9637cb..89f59bde0 100644
--- a/docker/fluxmq/node3.yaml
+++ b/docker/fluxmq/node3.yaml
@@ -140,7 +140,7 @@ queues:
 auth:
   # Principals admitted on the service listener. The certificate_uri_san must
   # match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
-  # the secret file must hold the same value as MG_RE_BROKER_SECRET.
+  # each secret file must hold the value configured by its service.
   local_principals:
     - name: "rules-engine"
       certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
@@ -161,6 +161,24 @@ auth:
         # The message stream the Rules Engine consumes.
         subscribe:
           - "m"
+    - name: "timescale-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
+    - name: "postgres-writer"
+      certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
+      role: "service"
+      current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
+      permissions:
+        # The writer is a passive consumer of the broker-provisioned stream. It
+        # neither publishes messages nor manages queues.
+        subscribe:
+          - "writers"
   external:
     url: "http://fluxmq-auth:7016"
     transport: "grpc"
diff --git a/docker/ssl/Makefile b/docker/ssl/Makefile
index 37d8c2773..ccadc4b76 100644
--- a/docker/ssl/Makefile
+++ b/docker/ssl/Makefile
@@ -51,6 +51,12 @@ FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME=fluxmq-service-server
 RE_FLUXMQ_CLIENT_CN=rules-engine
 RE_FLUXMQ_CLIENT_CRT_FILE_NAME=re-fluxmq-client
 RE_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/rules-engine
+TIMESCALE_WRITER_FLUXMQ_CLIENT_CN=timescale-writer
+TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=timescale-writer-fluxmq-client
+TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/timescale-writer
+POSTGRES_WRITER_FLUXMQ_CLIENT_CN=postgres-writer
+POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=postgres-writer-fluxmq-client
+POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/postgres-writer
 
 define GRPC_CERT_CONFIG
 [req]
@@ -131,7 +137,7 @@ It can be downloaded from $(DOWNLOAD_URL).
 
 etc, etc.
 endef
-all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs fluxmq_service_certs fluxmq_service_secret re_trace_key
+all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs fluxmq_service_certs fluxmq_service_secret timescale_writer_fluxmq_service_secret postgres_writer_fluxmq_service_secret re_trace_key
 
 # CA name and key is "ca".
 ca:
@@ -245,31 +251,53 @@ define gen_uri_san_cert
 endef
 
 # Certificates for the FluxMQ service listener: a server certificate for the
-# broker and a client certificate carrying the Rules Engine's URI SAN. The
+# broker and a client certificate carrying each local principal's URI SAN. The
 # listener requires mTLS, so both sides are issued by the same CA.
 fluxmq_service_certs:
 	$(call gen_fluxmq_server_cert,$(FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME))
 	$(call gen_uri_san_cert,$(RE_FLUXMQ_CLIENT_CRT_FILE_NAME),$(RE_FLUXMQ_CLIENT_CN),$(RE_FLUXMQ_CLIENT_URI_SAN))
+	$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
+	$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
 
-# The SASL secret the Rules Engine presents alongside its certificate. FluxMQ
-# reads it from a file and the Rules Engine sends it in its broker URL, so both
-# sides must see the same value: .env is the single source and this target
-# derives the file from it. FluxMQ requires at least 32 bytes.
-fluxmq_service_secret:
+# Generate a single principal's certificate when adding it to an existing
+# deployment, without rotating the broker certificate or the other principals.
+timescale_writer_fluxmq_client_cert:
+	$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
+
+postgres_writer_fluxmq_client_cert:
+	$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
+
+# The SASL secret a local principal presents alongside its certificate. FluxMQ
+# reads it from a file and the service sends it in its broker URL, so both sides
+# must see the same value: .env is the single source and these targets derive
+# the file from it. FluxMQ requires at least 32 bytes.
+#
+# Usage: $(call gen_fluxmq_service_secret,env_variable_name,secret_file_name)
+define gen_fluxmq_service_secret
 	@mkdir -p ../fluxmq/secrets
-	@secret="$${MG_RE_BROKER_SECRET}"; \
+	@secret="$${$(1)}"; \
 	if [ -z "$$secret" ]; then \
-		secret=$$(grep -E "^MG_RE_BROKER_SECRET=" ../.env | cut -d= -f2-); \
+		secret=$$(grep -E "^$(1)=" ../.env | cut -d= -f2-); \
 	fi; \
 	if [ -z "$$secret" ]; then \
-		echo "MG_RE_BROKER_SECRET is not set in docker/.env"; exit 1; \
+		echo "$(1) is not set in docker/.env"; exit 1; \
 	fi; \
 	if [ $${#secret} -lt 32 ]; then \
-		echo "MG_RE_BROKER_SECRET must be at least 32 characters"; exit 1; \
+		echo "$(1) must be at least 32 characters"; exit 1; \
 	fi; \
-	printf '%s' "$$secret" > ../fluxmq/secrets/re-current; \
-	chmod 600 ../fluxmq/secrets/re-current; \
-	echo "wrote docker/fluxmq/secrets/re-current"
+	printf '%s' "$$secret" > ../fluxmq/secrets/$(2); \
+	chmod 600 ../fluxmq/secrets/$(2); \
+	echo "wrote docker/fluxmq/secrets/$(2)"
+endef
+
+fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_RE_BROKER_SECRET,re-current)
+
+timescale_writer_fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_TIMESCALE_WRITER_BROKER_SECRET,timescale-writer-current)
+
+postgres_writer_fluxmq_service_secret:
+	$(call gen_fluxmq_service_secret,MG_POSTGRES_WRITER_BROKER_SECRET,postgres-writer-current)
 
 # Persistent HMAC key shared by Rules Engine replicas. Do not replace an
 # existing key: doing so while old messages are in flight would invalidate
diff --git a/go.mod b/go.mod
index d20c02f70..aee8a16c8 100644
--- a/go.mod
+++ b/go.mod
@@ -7,7 +7,7 @@ require (
 	connectrpc.com/otelconnect v0.9.0
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.18.2
-	github.com/absmach/fluxmq v0.50.0
+	github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62
 	github.com/absmach/senml v1.0.8
 	github.com/caarlos0/env/v10 v10.0.0
 	github.com/caarlos0/env/v11 v11.4.1
diff --git a/go.sum b/go.sum
index dddeee60a..507d5b917 100644
--- a/go.sum
+++ b/go.sum
@@ -20,8 +20,8 @@ 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.18.2 h1:dmopRHm2qTheHN1hdUKRRYpKwRrj7X9d8AWCFrb+K6s=
 github.com/absmach/callhome v0.18.2/go.mod h1:LEXKhES9JJtj3tBgTZv7VPNjOi5ukJQB0mFic0QP60Q=
-github.com/absmach/fluxmq v0.50.0 h1:RlYRv285ONSG6l0VqnFin7RLF8ZwAG2dkuHr6yn8iNo=
-github.com/absmach/fluxmq v0.50.0/go.mod h1:7sTd2lkxcq915f3Zp7mj1wXCBsuw9tdW+GT7GsLs8SI=
+github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62 h1:RhuJJ9vO3+6G2LzSqAKXaHBLRzhrE/C4tNq967AKWtw=
+github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62/go.mod h1:7sTd2lkxcq915f3Zp7mj1wXCBsuw9tdW+GT7GsLs8SI=
 github.com/absmach/senml v1.0.8 h1:+opem/r4g6c6eA/JLyCIuksyEhj7eBdysY3pEmy1mqo=
 github.com/absmach/senml v1.0.8/go.mod h1:DRhzHLgvQoIUHroBgpFrSWso+bJZO9E96RlHAHy+VRI=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
diff --git a/pkg/messaging/writers/brokers_fluxmq.go b/pkg/messaging/writers/brokers_fluxmq.go
index 728b9cabe..ddc4a9444 100644
--- a/pkg/messaging/writers/brokers_fluxmq.go
+++ b/pkg/messaging/writers/brokers_fluxmq.go
@@ -34,8 +34,23 @@ var cfg = jetstream.StreamConfig{
 	Storage:           jetstream.FileStorage,
 }
 
-func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
-	pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg), broker.ConnectionName("writers-msg-pubsub"))
+// InternalMetadata returns an option for a trusted local-service connection
+// that carries internal metadata over mTLS and consumes the broker-provisioned
+// writers stream. It lives here rather than being taken from
+// pkg/messaging/brokers because that package selects its backend on a different
+// build tag: an untagged build would pair a FluxMQ option with a NATS PubSub.
+func InternalMetadata(certFile, keyFile, caFile string) messaging.Option {
+	return broker.InternalMetadata(certFile, keyFile, caFile)
+}
+
+func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+		broker.ConnectionName("writers-msg-pubsub"),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/messaging/writers/brokers_nats.go b/pkg/messaging/writers/brokers_nats.go
index 50a1226be..c68d5ba21 100644
--- a/pkg/messaging/writers/brokers_nats.go
+++ b/pkg/messaging/writers/brokers_nats.go
@@ -34,8 +34,20 @@ var cfg = jetstream.StreamConfig{
 	Storage:           jetstream.FileStorage,
 }
 
-func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
-	pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg))
+// InternalMetadata is a no-op for the NATS backend. It exists for compile-time
+// compatibility with the FluxMQ variant; NATS carries metadata in the protobuf
+// message.
+func InternalMetadata(_, _, _ string) messaging.Option {
+	return func(_ any) error { return nil }
+}
+
+func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
+	brokerOpts := []messaging.Option{
+		broker.Prefix(prefix),
+		broker.JSStreamConfig(cfg),
+	}
+	brokerOpts = append(brokerOpts, opts...)
+	pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
 	if err != nil {
 		return nil, err
 	}