mqtt-5-learning/teaching/lessons/0005-contract-validation.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Lesson 5 — Validate the Message Before Acting</title>
  <link rel="stylesheet" href="../assets/lesson.css">
  <script src="../assets/quiz.js" defer></script>
</head>
<body>
  <main class="page">
    <p class="eyebrow">Go MQTT Vending Backend · Lesson 5 · 7 minutes</p>
    <h1>Validate the message before acting</h1>
    <p class="dek">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.</p>
    <div class="meta"><span class="pill">🎯 One skill: validate route and body together</span><span class="pill">🧠 Predict, then check</span><span class="pill">🧪 No broker required</span></div>

    <section class="card callout">
      <p class="question">❓ Start with the story</p>
      <p>A publication arrives on <code>lab/v1/devices/device-001/sales</code>. Its valid JSON body claims <code>device_id: device-002</code>. Should the backend accept it?</p>
      <p><strong>Say what could go wrong before revealing the model:</strong> the route and the body disagree about which machine owns this report.</p>
    </section>

    <h2>🗺️ First, see the full boundary</h2>
    <div class="flow" aria-label="message validation sequence">
      <div class="step">Broker delivers<br><strong>topic + bytes</strong></div><span class="arrow">→</span>
      <div class="step">Parse route<br><strong>device + kind</strong></div><span class="arrow">→</span>
      <div class="step">Decode contract<br><strong>shape + values</strong></div><span class="arrow">→</span>
      <div class="step">Only then<br><strong>application action</strong></div>
    </div>
    <p>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.</p>

    <h2>🔎 Read the code by responsibility</h2>
    <div class="two-col">
      <article class="card">
        <h3>📄 What the files do</h3>
        <p><code>internal/contracts/topic.go</code> defines the topic hierarchy, publish-name/filter types, builders, and parser. <code>message.go</code> defines payload structures and their checks. <code>decode.go</code> limits and decodes JSON, then invokes the correct contract validator.</p>
      </article>
      <article class="card">
        <h3>📥 `Decode` — short summary</h3>
        <p>Turns bytes into one typed message only after the size, JSON shape, topic identity, schema, timestamp, and message-specific rules pass.</p>
      </article>
    </div>
    <section class="card">
      <h3>🧭 Function overview: follow the boundary in order</h3>
      <p><strong><code>ParseTopic</code> summary:</strong> checks the shared <code>topicPrefix</code>, then validates the remaining device ID and message kind.</p>
      <p><strong><code>Decode[T]</code> summary:</strong> checks payload size and JSON shape, decodes one typed value, then calls that value's <code>Validate</code> method.</p>
      <p><strong><code>Validate</code> summary:</strong> 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.</p>
      <h4>🔎 Details: each layer narrows what the next layer must trust</h4>
      <ol>
        <li><code>ParseTopic</code> 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.</li>
        <li><code>Decode</code> rejects oversized bytes before allocating/parsing JSON, disallows unknown object fields, and rejects a second trailing JSON value. Only then does it construct <code>T</code>.</li>
        <li><code>validateEnvelope</code>, called by each contract's <code>Validate</code>, checks schema version, safe IDs, topic/body identity agreement, and timestamp bounds.</li>
        <li>Each concrete <code>Validate</code> 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.</li>
      </ol>
      <p>The backend entrypoint must stop on any returned error; only the typed value returned by successful <code>Decode</code> is eligible for application logic. Authentication and authorization still come from the broker/client configuration, not these payload checks.</p>
    </section>
    <article class="card">
      <h3>🧩 Details: route first, decode second</h3>
      <pre><code>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)</code></pre>
      <p><strong>Why this order?</strong> 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.</p>
      <p class="small">This focused fragment shows the order for the validation-stage handler. The current <code>cmd/backend/main.go</code> still logs raw payloads; the complete target entrypoints and contract files are included in the full-code section below.</p>
    </article>

    <h2>🧪 Check your model</h2>
    <form class="card quiz" data-quiz data-correct="b" data-explanation="Right. The topic supplies routing scope, and the backend still validates the body against that route before using it." data-retry="Separate the broker's routing choice from what the JSON body claims.">
      <fieldset class="question-block">
        <legend><strong>1. What do topic and payload prove together?</strong></legend>
        <label class="option"><input type="radio" name="q1" value="a"> Payload ID is only an untrusted claim</label>
        <label class="option"><input type="radio" name="q1" value="b"> Topic gives route; validate the body too</label>
        <label class="option"><input type="radio" name="q1" value="c"> Valid JSON proves no sender device identity</label>
      </fieldset>
      <button type="submit">Check answer</button><p class="feedback" data-feedback aria-live="polite"></p>
    </form>
    <form class="card quiz" data-quiz data-correct="c" data-explanation="Correct. The app limit checks payload bytes before JSON decoding; it is not a cap on the complete MQTT packet." data-retry="Which bytes does contracts.Decode receive? What packet bytes were already handled by Paho?">
      <fieldset class="question-block">
        <legend><strong>2. What does the 16 KiB application limit bound?</strong></legend>
        <label class="option"><input type="radio" name="q2" value="a"> Payload limit caps the full MQTT packet</label>
        <label class="option"><input type="radio" name="q2" value="b"> The broker rejects all oversized packet fields</label>
        <label class="option"><input type="radio" name="q2" value="c"> Application cap checks payload bytes before decoding</label>
      </fieldset>
      <button type="submit">Check answer</button><p class="feedback" data-feedback aria-live="polite"></p>
    </form>
    <form class="card quiz" data-quiz data-correct="c" data-explanation="Yes. MQTT transport acknowledgement and application validation are different events; a returned application error does not itself ask the broker to retry." data-retry="Does an application error change a packet exchange that the MQTT client already acknowledged automatically?">
      <fieldset class="question-block">
        <legend><strong>3. What can an invalid message still receive?</strong></legend>
        <label class="option"><input type="radio" name="q3" value="a"> Returning handler errors automatically requests broker retry</label>
        <label class="option"><input type="radio" name="q3" value="b"> Validation failure alone makes MQTT PUBACK impossible</label>
        <label class="option"><input type="radio" name="q3" value="c"> MQTT delivery and application validation are separate</label>
      </fieldset>
      <button type="submit">Check answer</button><p class="feedback" data-feedback aria-live="polite"></p>
    </form>

    <h2>🛠️ Your tiny practice</h2>
    <div class="card">
      <p>Predict which check rejects each input before looking at the implementation:</p>
      <ol>
        <li>Correct sales topic, body has an unknown field.</li>
        <li>Correct topic, valid JSON, body names a different device.</li>
        <li>Correct topic and identity, but one sale line has quantity zero.</li>
      </ol>
      <p>Write the validation stage for each: <em>topic parse</em>, <em>JSON/schema</em>, <em>shared identity</em>, or <em>message-specific value</em>. Then compare with <a href="../course-source/05-contracts-validation.md">the complete Section 5 contract code ↗</a>.</p>
    </div>

    
  <details class="card"><summary><strong>📦 Section 5 complete implementation files</strong></summary><p>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.</p><article class="card"><h3>📄 internal/contracts/topic.go</h3><pre><code>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) &gt; 128 {
		return false
	}
	for _, r := range value {
		if !(r &gt;= 'a' &amp;&amp; r &lt;= 'z' || r &gt;= 'A' &amp;&amp; r &lt;= 'Z' || r &gt;= '0' &amp;&amp; r &lt;= '9' || strings.ContainsRune("._-", r)) {
			return false
		}
	}
	return true
}
</code></pre></article><article class="card"><h3>📄 internal/contracts/message.go</h3><pre><code>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 &amp;&amp; (*m.Temperature &lt; -50 || *m.Temperature &gt; 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" &amp;&amp; m.Status != "completed" &amp;&amp; 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) &gt; 100 {
		return fmt.Errorf("sales report must contain 1 to 100 items")
	}
	for i, item := range m.Items {
		if !validID(item.SKU) || item.Quantity &lt;= 0 || item.AmountMinor &lt; 0 {
			return fmt.Errorf("item %d has invalid SKU, quantity, or amount_minor", i)
		}
	}
	return nil
}
</code></pre></article>
<article class="card"><h3>📄 internal/contracts/decode.go</h3><pre><code>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) &gt; MaxPayloadBytes {
		return value, fmt.Errorf("payload exceeds %d bytes", MaxPayloadBytes)
	}
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&amp;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(&amp;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
}
</code></pre></article>
<article class="card"><h3>📄 internal/contracts/decode_test.go</h3><pre><code>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")
	}
}
</code></pre></article>
<article class="card"><h3>📄 cmd/backend/main.go — validation-stage target</h3><pre><code>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 &lt;-ctx.Done():
			return nil
		case message := &lt;-inbox:
			if err := handleMessage(message); err != nil {
				log.Printf("rejected topic=%q reason=%v", message.Topic, err)
			}
		case &lt;-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
}
</code></pre></article>
<article class="card"><h3>📄 cmd/simulator/main.go — contract-stage target</h3><pre><code>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 &lt;-ctx.Done():
			return nil
		case &lt;-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 := &lt;-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)
}
</code></pre></article></details>

  <h2>🏗️ Build and run the validation boundary</h2>
  <p>After saving the contract files and updating both process entrypoints, run these from the repository root:</p>
  <pre><code>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</code></pre>
  <p>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.</p>
  <p>Run the simulator and backend in separate terminals:</p>
  <pre><code>/tmp/mqtt5-lab-simulator -username simulator -client-id lab-simulator-device-001</code></pre>
  <pre><code>/tmp/mqtt5-lab-backend -username backend -client-id lab-backend</code></pre>
  <p>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 <code>docker compose -p mqtt5-learning stop broker</code>. 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.</p>

  <section class="card callout"><h2>✅ Takeaway</h2><p><strong>Route identifies a topic scope; contract validation checks a payload claim; broker ACLs decide who may publish.</strong> Pass only a validated typed value to application work.</p><p>Ask me a follow-up if any of those boundaries still blur together.</p></section>
    <footer>
      <p>Start: <a href="../course-source/04-publish-subscribe-simulator.md">Section 4 source lesson</a> · Next: <a href="0005a-backend-message-processing.html">Lesson 5A — apply valid messages</a> · Reference: <a href="../reference/message-processing-lifecycle.html">message-processing lifecycle</a></p>
      <p>Primary sources: <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">OASIS MQTT 5.0 Standard</a>; <a href="https://pkg.go.dev/encoding/json">Go `encoding/json` documentation</a>. See <a href="../RESOURCES.md">workspace resources</a>.</p>
    </footer>
  </main>
</body>
</html>