Go MQTT Vending Backend · Lesson 5 · 7 minutes

Validate the message before acting

A broker-routed message still contains untrusted bytes. Build the boundary that stops a wrong or malformed vending report before it can change backend state.

🎯 One skill: validate route and body together🧠 Predict, then check🧪 No broker required

❓ Start with the story

A publication arrives on lab/v1/devices/device-001/sales. Its valid JSON body claims device_id: device-002. Should the backend accept it?

Say what could go wrong before revealing the model: the route and the body disagree about which machine owns this report.

🗺️ First, see the full boundary

Broker delivers
topic + bytes
→
Parse route
device + kind
→
Decode contract
shape + values
→
Only then
application action

The topic tells the backend where the broker routed the message. The payload claims what the message means. The backend must compare them; neither a matching topic nor valid JSON authenticates the sender. Mosquitto credentials and ACLs are a separate access-control boundary.

🔎 Read the code by responsibility

📄 What the files do

internal/contracts/topic.go defines the topic hierarchy, publish-name/filter types, builders, and parser. message.go defines payload structures and their checks. decode.go limits and decodes JSON, then invokes the correct contract validator.

📥 `Decode` — short summary

Turns bytes into one typed message only after the size, JSON shape, topic identity, schema, timestamp, and message-specific rules pass.

🧭 Function overview: follow the boundary in order

ParseTopic summary: checks the shared topicPrefix, then validates the remaining device ID and message kind.

Decode[T] summary: checks payload size and JSON shape, decodes one typed value, then calls that value's Validate method.

Validate summary: confirms that the message type matches the topic kind, shared envelope fields are valid, topic and body device IDs agree, and message-specific values are allowed.

🔎 Details: each layer narrows what the next layer must trust

  1. ParseTopic checks the configured prefix first, then splits and validates only the two remaining levels. Changing the shared prefix does not require rewriting the parser's level checks. It rejects malformed routes before selecting a payload type and does not prove which client sent the packet.
  2. Decode rejects oversized bytes before allocating/parsing JSON, disallows unknown object fields, and rejects a second trailing JSON value. Only then does it construct T.
  3. validateEnvelope, called by each contract's Validate, checks schema version, safe IDs, topic/body identity agreement, and timestamp bounds.
  4. Each concrete Validate method adds its own checks: telemetry temperature, supported command, response status, or sales-line values. A structurally valid message can still fail these domain rules.

The backend entrypoint must stop on any returned error; only the typed value returned by successful Decode is eligible for application logic. Authentication and authorization still come from the broker/client configuration, not these payload checks.

🧩 Details: route first, decode second

topic, err := contracts.ParseTopic(message.Topic)
if err != nil {
    return err // malformed route: no body handler runs
}

report, err := contracts.Decode[contracts.SalesReport](
    message.Payload, topic, time.Now().UTC(),
)
if err != nil {
    return err // bad shape, identity, schema, time, or values
}

log.Printf("valid sales report device=%s report=%s", report.DeviceID, report.ReportID)

Why this order? The parsed topic chooses the expected contract type. `Decode` rejects oversized input before parsing JSON, rejects unknown fields and extra JSON values, and calls `Validate`. Only a valid typed value reaches application handling. This section logs the validated report; Section 13 later persists sales.

This focused fragment shows the order for the validation-stage handler. The current cmd/backend/main.go still logs raw payloads; the complete target entrypoints and contract files are included in the full-code section below.

🧪 Check your model

1. What do topic and payload prove together?
2. What does the 16 KiB application limit bound?
3. What can an invalid message still receive?

🛠️ Your tiny practice

Predict which check rejects each input before looking at the implementation:

  1. Correct sales topic, body has an unknown field.
  2. Correct topic, valid JSON, body names a different device.
  3. Correct topic and identity, but one sale line has quantity zero.

Write the validation stage for each: topic parse, JSON/schema, shared identity, or message-specific value. Then compare with the complete Section 5 contract code ↗.

📦 Section 5 complete implementation files

Complete source files for this lesson are included here. The code blocks earlier in the lesson focus on the behavior being taught; use these full files when implementing it.

📄 internal/contracts/topic.go

package contracts

import (
	"fmt"
	"strings"
)

// TopicName is a concrete MQTT topic used by PUBLISH. It never contains MQTT
// subscription wildcards.
type TopicName string

// TopicFilter is an MQTT subscription pattern. This lab deliberately supports
// one device-level wildcard rather than accepting arbitrary filters.
type TopicFilter string

// Kind identifies the message flow in the final topic level.
type Kind string

const (
	KindTelemetry Kind = "telemetry"
	KindCommands  Kind = "commands"
	KindResponses Kind = "responses"
	KindSales     Kind = "sales"
)

const topicPrefix = "lab/v1/devices/"

// Topic is the parsed device identity and flow from a concrete topic name.
type Topic struct {
	DeviceID string
	Kind     Kind
}

func (t TopicName) String() string   { return string(t) }
func (f TopicFilter) String() string { return string(f) }

// DeviceTopic builds the concrete topic name for one device and supported flow.
func DeviceTopic(deviceID string, kind Kind) (TopicName, error) {
	if !validID(deviceID) {
		return "", fmt.Errorf("invalid device ID %q", deviceID)
	}
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicName(topicPrefix + deviceID + "/" + string(kind)), nil
}

// DeviceFilter builds a subscription filter for one flow across lab devices.
func DeviceFilter(kind Kind) (TopicFilter, error) {
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicFilter(topicPrefix + "+/" + string(kind)), nil
}

// DeviceSubscription builds an exact-topic filter for one device and flow.
func DeviceSubscription(deviceID string, kind Kind) (TopicFilter, error) {
	topic, err := DeviceTopic(deviceID, kind)
	if err != nil {
		return "", err
	}
	return TopicFilter(topic.String()), nil
}

// ParseTopic turns broker routing metadata into values handlers can compare.
func ParseTopic(value string) (Topic, error) {
	if !strings.HasPrefix(value, topicPrefix) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	parts := strings.Split(strings.TrimPrefix(value, topicPrefix), "/")
	if len(parts) != 2 || !validID(parts[0]) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	kind := Kind(parts[1])
	if !supportedKind(kind) {
		return Topic{}, fmt.Errorf("unsupported topic kind %q", kind)
	}
	return Topic{DeviceID: parts[0], Kind: kind}, nil
}

func supportedKind(kind Kind) bool {
	switch kind {
	case KindTelemetry, KindCommands, KindResponses, KindSales:
		return true
	default:
		return false
	}
}

// validID limits identifiers to safe, single-segment topic values.
func validID(value string) bool {
	if value == "" || len(value) > 128 {
		return false
	}
	for _, r := range value {
		if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r)) {
			return false
		}
	}
	return true
}

📄 internal/contracts/message.go

package contracts

import (
	"fmt"
	"time"
)

const SchemaVersion = 1

// MQTTMessage is implemented by each versioned message contract.
// Validation uses the parsed route and caller-supplied time for deterministic checks.
type MQTTMessage interface {
	Validate(topic Topic, now time.Time) error
}

// These types define the JSON fields shared by backend and simulator.
type Heartbeat struct {
	SchemaVersion int      `json:"schema_version"`
	MessageID     string   `json:"message_id"`
	DeviceID      string   `json:"device_id"`
	OccurredAt    string   `json:"occurred_at"`
	Temperature   *float64 `json:"temperature,omitempty"`
}

type Command struct {
	SchemaVersion int    `json:"schema_version"`
	MessageID     string `json:"message_id"`
	DeviceID      string `json:"device_id"`
	OccurredAt    string `json:"occurred_at"`
	CommandID     string `json:"command_id"`
	Command       string `json:"command"`
}

type CommandResult struct {
	SchemaVersion int    `json:"schema_version"`
	MessageID     string `json:"message_id"`
	DeviceID      string `json:"device_id"`
	OccurredAt    string `json:"occurred_at"`
	CommandID     string `json:"command_id"`
	Status        string `json:"status"`
	Detail        string `json:"detail,omitempty"`
}

type SalesLine struct {
	SKU         string `json:"sku"`
	Quantity    int64  `json:"quantity"`
	AmountMinor int64  `json:"amount_minor"`
}

type SalesReport struct {
	SchemaVersion int         `json:"schema_version"`
	ReportID      string      `json:"report_id"`
	DeviceID      string      `json:"device_id"`
	OccurredAt    string      `json:"occurred_at"`
	Items         []SalesLine `json:"items"`
}

// validateEnvelope checks fields shared by every versioned message.
func validateEnvelope(topic Topic, version int, messageID, deviceID, occurredAt string, now time.Time) error {
	if version != SchemaVersion {
		return fmt.Errorf("unsupported schema_version %d", version)
	}
	if !validID(messageID) || !validID(deviceID) {
		return fmt.Errorf("message_id and device_id must be valid IDs")
	}
	if topic.DeviceID != deviceID {
		return fmt.Errorf("topic device %q does not match payload device_id %q", topic.DeviceID, deviceID)
	}
	stamp, err := time.Parse(time.RFC3339, occurredAt)
	if err != nil {
		return fmt.Errorf("occurred_at must be RFC3339: %w", err)
	}
	if stamp.After(now.Add(30 * time.Second)) {
		return fmt.Errorf("occurred_at is more than 30 seconds in the future")
	}
	return nil
}

// Each Validate method adds checks specific to one message kind.
func (m Heartbeat) Validate(topic Topic, now time.Time) error {
	if topic.Kind != KindTelemetry {
		return fmt.Errorf("heartbeat requires telemetry topic")
	}
	if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil {
		return err
	}
	if m.Temperature != nil && (*m.Temperature < -50 || *m.Temperature > 100) {
		return fmt.Errorf("temperature is outside the supported range")
	}
	return nil
}

func (m Command) Validate(topic Topic, now time.Time) error {
	if topic.Kind != KindCommands {
		return fmt.Errorf("command requires commands topic")
	}
	if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil {
		return err
	}
	if !validID(m.CommandID) || m.Command != "ping" {
		return fmt.Errorf("command_id is invalid or command is unsupported")
	}
	return nil
}

func (m CommandResult) Validate(topic Topic, now time.Time) error {
	if topic.Kind != KindResponses {
		return fmt.Errorf("command result requires responses topic")
	}
	if err := validateEnvelope(topic, m.SchemaVersion, m.MessageID, m.DeviceID, m.OccurredAt, now); err != nil {
		return err
	}
	if !validID(m.CommandID) || (m.Status != "accepted" && m.Status != "completed" && m.Status != "rejected") {
		return fmt.Errorf("command result has invalid command_id or status")
	}
	return nil
}

func (m SalesReport) Validate(topic Topic, now time.Time) error {
	if topic.Kind != KindSales {
		return fmt.Errorf("sales report requires sales topic")
	}
	if err := validateEnvelope(topic, m.SchemaVersion, m.ReportID, m.DeviceID, m.OccurredAt, now); err != nil {
		return err
	}
	if len(m.Items) == 0 || len(m.Items) > 100 {
		return fmt.Errorf("sales report must contain 1 to 100 items")
	}
	for i, item := range m.Items {
		if !validID(item.SKU) || item.Quantity <= 0 || item.AmountMinor < 0 {
			return fmt.Errorf("item %d has invalid SKU, quantity, or amount_minor", i)
		}
	}
	return nil
}

📄 internal/contracts/decode.go

package contracts

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"time"
)

const MaxPayloadBytes = 16 * 1024

// Decode bounds input, rejects unknown fields, requires one JSON value,
// and validates it against the topic before returning it.
func Decode[T MQTTMessage](payload []byte, topic Topic, now time.Time) (T, error) {
	var value T
	if len(payload) > MaxPayloadBytes {
		return value, fmt.Errorf("payload exceeds %d bytes", MaxPayloadBytes)
	}
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&value); err != nil {
		return value, fmt.Errorf("decode JSON: %w", err)
	}
	var trailing any
	// Decode reads one JSON value at a time. A second call distinguishes
	// whitespace (EOF) from a second value or malformed trailing bytes.
	if err := decoder.Decode(&trailing); err != io.EOF {
		if err == nil {
			return value, fmt.Errorf("payload contains more than one JSON value")
		}
		return value, fmt.Errorf("trailing data after JSON value: %w", err)
	}
	if err := value.Validate(topic, now); err != nil {
		return value, fmt.Errorf("validate message: %w", err)
	}
	return value, nil
}

📄 internal/contracts/decode_test.go

package contracts

import (
	"strings"
	"testing"
	"time"
)

// Topic tests reject malformed routes and unsupported message kinds.
func TestParseTopic(t *testing.T) {
	topic, err := ParseTopic("lab/v1/devices/device-001/sales")
	if err != nil {
		t.Fatal(err)
	}
	if topic.DeviceID != "device-001" || topic.Kind != KindSales {
		t.Fatalf("unexpected topic: %#v", topic)
	}
	for _, invalid := range []string{"lab/v1/devices/+/sales", "lab/v1/devices/device-001/unknown", "lab/v1/devices/device-001/sales/extra"} {
		if _, err := ParseTopic(invalid); err == nil {
			t.Errorf("ParseTopic(%q) succeeded", invalid)
		}
	}
}

// These cases cover strict JSON shape and single-value decoding.
func TestDecodeRejectsUnknownAndTrailingValues(t *testing.T) {
	topic := Topic{DeviceID: "device-001", Kind: KindTelemetry}
	now := time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)
	if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"unknown":true}`), topic, now); err == nil {
		t.Fatal("unknown field was accepted")
	}
	if _, err := Decode[Heartbeat]([]byte(`{"schema_version":1,"message_id":"m-1","device_id":"device-001","occurred_at":"2026-10-13T12:00:00Z"} {"message_id":"m-2"}`), topic, now); err == nil {
		t.Fatal("trailing JSON value was accepted")
	}
}

// This case ensures the application limit is applied before decoding.
func TestDecodeRejectsOversizedPayload(t *testing.T) {
	payload := []byte(`"` + strings.Repeat("x", MaxPayloadBytes) + `"`)
	if _, err := Decode[Heartbeat](payload, topic, now); err == nil {
		t.Fatal("oversized payload was accepted")
	}
}

// A body cannot override the device selected by the topic.
func TestCommandRejectsTopicIdentityMismatch(t *testing.T) {
	topic, err := ParseTopic("lab/v1/devices/device-001/commands")
	if err != nil {
		t.Fatal(err)
	}
	command := Command{SchemaVersion: SchemaVersion, MessageID: "m-1", DeviceID: "device-002", OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: "c-1", Command: "ping"}
	if err := command.Validate(topic, time.Now().UTC()); err == nil {
		t.Fatal("topic/body identity mismatch was accepted")
	}
}

📄 cmd/backend/main.go — validation-stage target

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/mqtt"
)

func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return fmt.Errorf("load config: %w", err)
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	telemetryFilter, err := contracts.DeviceFilter(contracts.KindTelemetry)
	if err != nil { return err }
	responsesFilter, err := contracts.DeviceFilter(contracts.KindResponses)
	if err != nil { return err }
	salesFilter, err := contracts.DeviceFilter(contracts.KindSales)
	if err != nil { return err }
	commandTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	// Start subscriptions before handling any delivered message.
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{telemetryFilter, responsesFilter, salesFilter}, inbox)
	if err != nil {
		return fmt.Errorf("start MQTT: %w", err)
	}
	defer mqtt.Stop(manager)

	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	sequence := 0
	for {
		select {
		case <-ctx.Done():
			return nil
		case message := <-inbox:
			if err := handleMessage(message); err != nil {
				log.Printf("rejected topic=%q reason=%v", message.Topic, err)
			}
		case <-ticker.C:
			sequence++
			id := fmt.Sprintf("ping-%d-%d", time.Now().UTC().Unix(), sequence)
			command := contracts.Command{SchemaVersion: contracts.SchemaVersion, MessageID: id, DeviceID: cfg.MQTT.DeviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: id, Command: "ping"}
			payload, err := json.Marshal(command)
			if err != nil {
				return fmt.Errorf("encode command: %w", err)
			}
			if err := mqtt.Publish(ctx, manager, commandTopic, payload); err != nil {
				log.Printf("command publish failed or uncertain id=%s: %v", id, err)
			}
		}
	}
}

func handleMessage(message mqtt.Message) error {
	// Parse and validate before a message can reach application handling.
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	switch topic.Kind {
	case contracts.KindTelemetry:
		value, err := contracts.Decode[contracts.Heartbeat](message.Payload, topic, now)
		if err != nil {
			return err
		}
		log.Printf("valid heartbeat device=%s message=%s", value.DeviceID, value.MessageID)
	case contracts.KindResponses:
		value, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, now)
		if err != nil {
			return err
		}
		log.Printf("command result device=%s command=%s status=%s", value.DeviceID, value.CommandID, value.Status)
	case contracts.KindSales:
		value, err := contracts.Decode[contracts.SalesReport](message.Payload, topic, now)
		if err != nil {
			return err
		}
		log.Printf("valid sales report device=%s report=%s items=%d", value.DeviceID, value.ReportID, len(value.Items))
	default:
		return fmt.Errorf("backend does not consume topic kind %q", topic.Kind)
	}
	return nil
}

📄 cmd/simulator/main.go — contract-stage target

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/mqtt"
	"github.com/eclipse/paho.golang/autopaho"
)

func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return fmt.Errorf("load config: %w", err)
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	// Publish this device's heartbeat and subscribe only to its command topic.
	commandFilter, err := contracts.DeviceSubscription(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	telemetryTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindTelemetry)
	if err != nil { return err }
	responseTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindResponses)
	if err != nil { return err }
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{commandFilter}, inbox)
	if err != nil {
		return fmt.Errorf("start MQTT: %w", err)
	}
	defer mqtt.Stop(manager)
	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	sequence := 0
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			sequence++
			heartbeat := contracts.Heartbeat{SchemaVersion: contracts.SchemaVersion, MessageID: fmt.Sprintf("heartbeat-%d-%d", time.Now().Unix(), sequence), DeviceID: cfg.MQTT.DeviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339)}
			payload, err := json.Marshal(heartbeat)
			if err != nil {
				return fmt.Errorf("encode heartbeat: %w", err)
			}
			if err := mqtt.Publish(ctx, manager, telemetryTopic, payload); err != nil {
				log.Printf("heartbeat publish: %v", err)
			}
		case message := <-inbox:
			if message.Retain {
				log.Printf("reject retained command topic=%q", message.Topic)
				continue
			}
			if err := respond(ctx, manager, cfg.MQTT.DeviceID, message); err != nil {
				log.Printf("reject command topic=%q reason=%v", message.Topic, err)
			}
		}
	}
}

func respond(ctx context.Context, manager *autopaho.ConnectionManager, deviceID string, message mqtt.Message) error {
	// Check route and body identity before constructing a command result.
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return err
	}
	command, err := contracts.Decode[contracts.Command](message.Payload, topic, time.Now().UTC())
	if err != nil {
		return err
	}
	if topic.DeviceID != deviceID {
		return fmt.Errorf("command is for another device")
	}
	result := contracts.CommandResult{SchemaVersion: contracts.SchemaVersion, MessageID: "result-" + command.CommandID, DeviceID: deviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: command.CommandID, Status: "completed"}
	payload, err := json.Marshal(result)
	if err != nil {
		return fmt.Errorf("encode result: %w", err)
	}
	responseTopic, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return fmt.Errorf("build response topic: %w", err) }
	return mqtt.Publish(ctx, manager, responseTopic, payload)
}

🏗️ Build and run the validation boundary

After saving the contract files and updating both process entrypoints, run these from the repository root:

go test ./internal/contracts
go build -o /tmp/mqtt5-lab-backend ./cmd/backend
go build -o /tmp/mqtt5-lab-simulator ./cmd/simulator
docker compose -p mqtt5-learning up -d broker
docker compose -p mqtt5-learning ps

The contract tests exercise the topic and payload cases in the test file. Building each process checks that the typed contract is wired through the real programs. The broker command starts only the isolated learning service.

Run the simulator and backend in separate terminals:

/tmp/mqtt5-lab-simulator -username simulator -client-id lab-simulator-device-001
/tmp/mqtt5-lab-backend -username backend -client-id lab-backend

Expect the backend to log validated heartbeat identifiers and the simulator to remain connected. Then stop both clients with Ctrl+C and stop the broker with docker compose -p mqtt5-learning stop broker. This smoke run checks valid traffic; the unit tests cover malformed, oversized, and identity-mismatch inputs without relying on broker delivery. Ask me a follow-up if the result differs from your prediction.

✅ Takeaway

Route identifies a topic scope; contract validation checks a payload claim; broker ACLs decide who may publish. Pass only a validated typed value to application work.

Ask me a follow-up if any of those boundaries still blur together.