A base YAML file defines shared defaults (transports, processors, stores). A DynamoDB overlay adds environment-specific settings (queue URLs, broker addresses, deployment mode). This separates concerns: developers manage the base config in version control, operations manage the overlay in DynamoDB.
Your organization deploys the same bridge across development, staging, and production environments. The routing logic, processor chain, and structural config are identical everywhere – only infrastructure endpoints, credentials, and scaling parameters differ. You want developers to own the base configuration in git where it can be reviewed and versioned, while operations teams manage environment-specific overrides through a centralized DynamoDB table without requiring code changes or redeployments.
The layered configuration pattern solves this by composing a final BridgeConfig from multiple sources. The base layer provides sensible defaults. The DynamoDB overlay patches in environment-specific values. The Manager merges them, validates the result, and emits configuration change events when either layer updates.
flowchart TD
subgraph Sources ["Configuration Sources"]
File["Base Layer\n(base.yaml in git)"]
DDB["DynamoDB Overlay\n(per-environment)"]
end
File --> Mgr["config.Manager"]
DDB --> Mgr
Mgr --> Merge["DefaultMerge"]
Merge --> Validate["config.Validate()"]
Validate --> Build["bridge.Builder.Build()"]
Build --> RT["Running Bridge"]
DDB -.->|"PollInterval: 30s"| Mgr
File -.->|"fsnotify / poll"| Mgr
RT -.->|"watch channel"| Mgr
style RT fill:#2d6,stroke:#333
style Merge fill:#f96,stroke:#333
Both sources can emit change notifications. The file watcher uses fsnotify or SHA-256 polling. The DynamoDB loader uses version-number polling. When either source changes, the Manager re-merges all layers, validates the result, and sends the new config through the watch channel.
This file lives in the repository and defines the structural skeleton – sessions, receivers, senders, bindings, routes, and development-safe store backends. It deliberately carries no broker URL – the overlay below supplies it – so on its own it is not a configuration a bridge can be built from.
# base.yaml -- checked into git
bridge:
id: my-bridge
shutdown_timeout: 30s
config_watch:
mode: notify
debounce: 200ms
sessions:
- id: mqtt-conn
transport: mqtt
options:
session:
client_id: bridge-01
keep_alive: 30
connect_timeout: 30s
stores:
outbox:
type: memory
options:
acknowledge_volatile: true
dlq:
type: memory
options:
acknowledge_volatile: true
receivers:
- id: telemetry-in
session_id: mqtt-conn
topics:
- topic: "telemetry/#"
qos: 1
senders:
- id: sqs-out
transport: sqs
options:
region: us-west-1
batch_size: 10
bindings:
- id: to-sqs
sender_id: sqs-out
address: telemetry-events
routes:
- id: ingest
receiver_id: telemetry-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-sqs]
policy:
max_in_flight: 50
on_permanent_failure: dlq
This config works standalone for local development – the MQTT broker URL defaults to localhost, stores use in-memory backends, and the deployment mode is standalone (the default). No secrets, no cloud infrastructure.
The overlay is stored as a single DynamoDB item. Only the fields that differ from the base need to be present. Missing fields are inherited from the base layer.
{
"bridge": {
"deployment_mode": "clustered",
"instance_id": "prod-01"
},
"sessions": [
{
"id": "mqtt-conn",
"session_mode": "exclusive",
"options": {
"session": {
"broker_url": "tls://mqtt.prod.example.com:8883",
"client_id": "prod-bridge-01",
"tls": {
"enable": true,
"ca_cert_file": "/etc/certs/ca.pem"
}
}
}
}
],
"senders": [
{
"id": "sqs-out",
"options": {
"queue_url": "https://sqs.us-west-1.amazonaws.com/123456789/telemetry-events"
}
}
],
"stores": {
"lease": {
"type": "dynamodb",
"options": { "table_name": "prod-leases" }
},
"outbox": {
"type": "dynamodb",
"options": { "table_name": "prod-outbox" }
},
"dlq": {
"type": "dynamodb",
"options": { "table_name": "prod-dlq" }
}
},
"routes": [
{
"id": "ingest",
"delivery_mode": "shared_outbox",
"policy": {
"max_in_flight": 200,
"ack_after": "outbox_persist"
},
"session": {
"session_id": "mqtt-conn",
"sender_id": "sqs-out",
"lease_ttl": "300s",
"connect_after_lease": true
}
}
]
}
The overlay transforms a simple local-dev config into a production-grade clustered deployment. It adds the production broker URL, switches stores from memory to DynamoDB, enables exclusive sessions with lease coordination, and increases concurrency.
package main
import (
"context"
"log/slog"
"os"
"os/signal"
"time"
awsconfig "github.com/aws/aws-sdk-go-v2/config"
awsdynamodb "github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/mariotoffia/gobridge/bridge"
"github.com/mariotoffia/gobridge/config"
fileconfig "github.com/mariotoffia/gobridge/adapters/native/config/file"
ddbconfig "github.com/mariotoffia/gobridge/adapters/aws/config/dynamodb"
nativestore "github.com/mariotoffia/gobridge/adapters/native/store"
awsstore "github.com/mariotoffia/gobridge/adapters/aws/store"
paho "github.com/mariotoffia/gobridge/adapters/mqtt/transport/paho"
sqs "github.com/mariotoffia/gobridge/adapters/aws/transport/sqs"
"github.com/mariotoffia/gobridge/ports"
)
func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
logger := slog.Default()
// --- Registry: register the adapter decoders this config references.
// NewSource and NewWatcher both require the *ports.Registry. ---
reg := ports.NewRegistry()
_ = paho.Register(reg)
_ = sqs.Register(reg)
_ = nativestore.Register(reg)
// --- Base layer: file source with watcher (both take the registry) ---
fileSource := fileconfig.NewSource("base.yaml", reg)
fileWatcher := fileconfig.NewWatcher("base.yaml", reg,
fileconfig.WithLogger(logger),
)
// --- AWS SDK client, shared by the DynamoDB config loader and store
// factory. Mirrors the wiring in cmd/gobridge/main.go. ---
awsCfg, err := awsconfig.LoadDefaultConfig(ctx)
if err != nil {
slog.Error("aws config load failed", "error", err)
os.Exit(1)
}
ddbClient := awsdynamodb.NewFromConfig(awsCfg)
// --- Overlay layer: DynamoDB source with polling ---
ddbLoader := ddbconfig.NewLoader(ddbClient,
ddbconfig.WithTableName("gobridge-config"),
ddbconfig.WithBridgeID("production"),
ddbconfig.WithPollInterval(30 * time.Second),
)
// --- Compose layers into a Manager ---
mgr := config.NewManager(
config.Layer{Name: "base", Loader: fileSource, Watcher: fileWatcher},
config.WithOverlay(config.Layer{
Name: "env", Loader: ddbLoader, Watcher: ddbLoader,
}),
config.WithManagerLogger(logger),
)
// Load merges base + overlay, then validates
cfg, err := mgr.Load(ctx)
if err != nil {
slog.Error("config load failed", "error", err)
os.Exit(1)
}
// Watch emits merged configs when either layer changes
watchCh, err := mgr.Watch(ctx)
if err != nil {
slog.Error("config watch failed", "error", err)
os.Exit(1)
}
defer mgr.Stop()
// --- Build and run the bridge ---
sup := bridge.NewSupervisor(
bridge.WithSupervisorLogger(logger),
bridge.WithReconfigStrategy(
bridge.NewWindowedStrategy(10*time.Second, 30*time.Second, nil),
),
)
sup.RegisterTransport("mqtt", paho.NewFactory(logger))
sup.RegisterTransport("sqs", sqs.NewFactory(logger))
sup.RegisterStoreFactory("memory", nativestore.NewMemoryStoreFactory())
sup.RegisterStoreFactory("dynamodb", awsstore.NewDynamoDBStoreFactory(ddbClient))
// Run blocks until ctx is cancelled
sup.Run(ctx, cfg, watchCh)
}
The Manager handles all merge and validation logic internally. The caller receives a fully merged, validated BridgeConfig from Load() and a channel of subsequent merged configs from Watch().
The DefaultMerge function applies overlay values onto the base config using field-level and ID-level merge rules. Understanding these rules is essential for predicting the final config.
bridge settings: non-zero field replacementEach non-zero field in the overlay replaces the corresponding base field. Zero-value fields in the overlay are ignored (they inherit the base value).
| Field | Base | Overlay | Merged |
|---|---|---|---|
id |
my-bridge |
(not set) | my-bridge |
deployment_mode |
(default: standalone) | clustered |
clustered |
instance_id |
(not set) | prod-01 |
prod-01 |
shutdown_timeout |
30s |
(not set) | 30s |
sessions, receivers, senders, bindings, routes: merge by IDCollections are merged by matching the id field. When an overlay entry has the same ID as a base entry, the overlay replaces the base entry entirely. When an overlay entry has a new ID not present in the base, it is appended.
flowchart LR
subgraph Base ["Base Layer"]
BS["mqtt-conn\n(no session.broker_url,\nephemeral)"]
end
subgraph Overlay ["DynamoDB Overlay"]
OS["mqtt-conn\n(session.broker_url: tls://...,\nexclusive)"]
end
subgraph Merged ["Merged Result"]
MS["mqtt-conn\n(session.broker_url: tls://...,\nexclusive,\nsession.client_id: prod-bridge-01)"]
end
BS -->|"ID match: field-merge"| MS
OS -->|"overlay wins"| MS
Important: When IDs match, scalar fields merge field-level – a non-empty overlay scalar (like session_mode) wins, and a scalar the overlay omits is inherited from the base. The typed plugin options map is not deep-merged: an overlay that supplies an options block replaces the base’s options for that entry wholesale, so it must include every required option (like client_id), not just the ones being changed. An overlay that omits options entirely carries the base options forward – this is how a scalar-only patch (e.g. only session_mode) avoids erasing broker URLs and credentials (see config.DefaultMerge).
In the example above, the overlay session includes both broker_url (new) and client_id: prod-bridge-01 (overriding the base value of bridge-01). If the overlay omitted client_id, the merged session would have no client_id and validation would fail.
stores: per-role replacementEach store role (lease, outbox, dlq) is treated independently. If the overlay defines a role, it replaces the base definition for that role. If the overlay does not define a role, the base definition is preserved.
| Role | Base | Overlay | Merged |
|---|---|---|---|
lease |
(nil) | dynamodb (prod-leases) |
dynamodb (prod-leases) |
outbox |
memory |
dynamodb (prod-outbox) |
dynamodb (prod-outbox) |
dlq |
memory |
dynamodb (prod-dlq) |
dynamodb (prod-dlq) |
config_watch, http: wholesale replacementThese top-level objects are replaced entirely if the overlay provides a non-nil value. If the overlay omits them, the base values are preserved.
In the example, the overlay does not include config_watch, so the base definition (mode: notify, debounce: 200ms) carries through unchanged.
If the overlay adds a receiver, sender, binding, or route with an ID not present in the base, it is appended to the collection:
{
"receivers": [
{ "id": "webhook-in", "transport": "http", "options": { "path": "/webhooks" } }
]
}
This overlay would result in two receivers: telemetry-in (from base) and webhook-in (from overlay).
The following diagram shows the complete merge result for this scenario.
flowchart TD
subgraph Base ["base.yaml"]
B_bridge["bridge:\n id: my-bridge\n shutdown_timeout: 30s"]
B_session["sessions:\n mqtt-conn (ephemeral)"]
B_stores["stores:\n outbox: memory\n dlq: memory"]
B_route["routes:\n ingest (direct_hold, mif=50)"]
end
subgraph Overlay ["DynamoDB Overlay"]
O_bridge["bridge:\n deployment_mode: clustered\n instance_id: prod-01"]
O_session["sessions:\n mqtt-conn (exclusive, TLS)"]
O_stores["stores:\n lease: dynamodb\n outbox: dynamodb\n dlq: dynamodb"]
O_route["routes:\n ingest (shared_outbox, mif=200)"]
end
subgraph Merged ["Merged Config"]
M_bridge["bridge:\n id: my-bridge\n deployment_mode: clustered\n instance_id: prod-01\n shutdown_timeout: 30s"]
M_session["sessions:\n mqtt-conn (exclusive, TLS,\n broker: tls://mqtt.prod...)"]
M_stores["stores:\n lease: dynamodb\n outbox: dynamodb\n dlq: dynamodb"]
M_route["routes:\n ingest (shared_outbox, mif=200,\n session: mqtt-conn)"]
end
B_bridge --> M_bridge
O_bridge --> M_bridge
B_session --> M_session
O_session --> M_session
B_stores --> M_stores
O_stores --> M_stores
B_route --> M_route
O_route --> M_route
style Merged fill:#2d6,stroke:#333
| Pattern | Best For | Advantages | Drawbacks |
|---|---|---|---|
| File-only | Simple deployments, all config in git | Auditable, no external deps | Requires file edit to change; no env separation |
| DynamoDB-only | Centralized config management | Centralized, environment-aware | No git history; requires DynamoDB everywhere |
| Layered (recommended) | Separation of concerns | Devs own structure in git; ops override in DynamoDB | Requires understanding merge semantics |
Add a third layer for instance-specific overrides. Layers are applied in order: base, then environment, then instance.
instanceSource := fileconfig.NewSource("/etc/gobridge/instance.yaml", reg)
instanceWatcher := fileconfig.NewWatcher("/etc/gobridge/instance.yaml", reg)
mgr := config.NewManager(
config.Layer{Name: "base", Loader: fileSource, Watcher: fileWatcher},
config.WithOverlay(config.Layer{Name: "env", Loader: ddbLoader, Watcher: ddbLoader}),
config.WithOverlay(config.Layer{
Name: "instance", Loader: instanceSource, Watcher: instanceWatcher,
}),
)
The instance-level overlay might contain only the instance_id and client_id:
# /etc/gobridge/instance.yaml
bridge:
instance_id: prod-03
sessions:
- id: mqtt-conn
options:
session:
client_id: prod-bridge-03
This overlay is a partial layer: it merges onto the base config above, so it omits transport and other required fields that the base already supplies. It does not decode on its own.
This three-layer approach is useful when you have many instances in a cluster and each needs a unique identity, but all share the same environment-level config.
The default merge behavior can be replaced with a custom function using WithMergeFunc. This is useful when you need non-standard merge logic, such as deep-merging options maps instead of replacing them.
mgr := config.NewManager(
config.Layer{Name: "base", Loader: fileSource, Watcher: fileWatcher},
config.WithOverlay(config.Layer{Name: "env", Loader: ddbLoader, Watcher: ddbLoader}),
config.WithMergeFunc(func(base, overlay *ports.BridgeConfig) (*ports.BridgeConfig, error) {
// Custom merge logic here
// For example, deep-merge session options instead of replacing
merged, err := config.DefaultMerge(base, overlay)
if err != nil {
return nil, err
}
// ... apply additional transformations ...
return merged, nil
}),
)
The custom merge function receives the base config and the next overlay. It is called once for each overlay in order. The return value becomes the new base for the next overlay. If the function returns an error, the Manager propagates it from Load() or through the watch channel.
The DynamoDB config table uses the following schema. One item per bridge, identified by the bridge ID:
| Attribute | Type | Key | Description |
|---|---|---|---|
PK |
String | Partition | config#<bridge-id> (e.g., config#production) |
SK |
String | Sort | current |
data |
String | – | The full or partial BridgeConfig serialized as a JSON string |
version |
Number | – | Monotonically increasing version for change detection and compare-and-set writes |
The DynamoDB loader detects changes via DynamoDB Streams by default (falling
back to version-number polling at PollInterval when streams are not
enabled on the table). On a change it reads the item with a strongly
consistent read, parses the data JSON string into a BridgeConfig, and
emits it through the watcher channel.
Reads are strongly consistent so an admin Save is observed immediately by
watchers. Save is a compare-and-set: it reads the current version, then
writes version+1 guarded by a ConditionExpression on the unchanged
version. Concurrent admin writes therefore cannot silently lose updates —
the losing writer receives a version-mismatch error and must reload and
retry.
To update the overlay out-of-band, write a new item whose data JSON string
carries the new config and whose version is incremented; guard the write on
the previous version to avoid clobbering a concurrent update.