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.

🗺️ First, see the full boundary

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

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)

🧪 Check your model

🛠️ Your tiny practice

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
}

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
}

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
}

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")
	}
}

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
}

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.