mqtt-5-learning/teaching/lessons/0012-vending-workflows.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Lesson 12 — Vending Workflows</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 12 · 12 minutes</p>
  <h1>Connect MQTT patterns to vending workflows</h1>
  <p class="dek">Apply stable business identities and explicit application evidence to sales, catalog, configuration, health, and device logs.</p>
  <div class="meta"><span class="pill">🎯 One skill: define workflow evidence</span><span class="pill">🧠 Predict duplicate and stale effects</span><span class="pill">🧪 Synthetic data only</span></div>
  <section class="card callout"><p class="question">❓ Start with the system story</p><p>A simulator sends the same sale report twice, receives a new product catalog, and later asks for configuration. Which side owns each fact, and what proves the device applied the catalog?</p><p>Pause: separate backend business records from device-reported state.</p></section>
  <h2>💰 Sales: stable identity survives retries</h2>
  <p>Deduplicate with a business key such as <code>(device_id, report_id)</code>. MQTT packet identifiers belong to protocol exchanges and can be reused; they are not sale IDs. In this lesson's in-memory stage, replay the same report without repeating its business effect, and state clearly that a backend restart forgets the deduplication map.</p>
  <h2>📦 Catalog: send intent, receive application evidence</h2>
  <div class="flow" aria-label="catalog delivery"><div class="step">Backend<br><strong>desired revision 9</strong></div><span class="arrow">→</span><div class="step">Broker<br><strong>PUBACK for QoS hop</strong></div><span class="arrow">→</span><div class="step">Simulator<br><strong>validate + apply</strong></div><span class="arrow">→</span><div class="step">Backend<br><strong>reported revision 9</strong></div></div>
  <p>Include schema version, stable message/job ID, and monotonic revision. Have the simulator reject revision 8 after applying revision 9. If you model catalog as replaceable desired state, retained state may fit; if each delivery is a one-time job, publish it nonretained. Pick one semantic model and explain it.</p>
  <form class="card quiz" data-quiz data-correct="c" data-explanation="The simulator's validated application response or reported revision is evidence that it applied the catalog. PUBACK only acknowledges the relevant MQTT delivery step." data-retry="Ask which process produced the evidence and whether it saw the application effect.">
    <p><strong>🧠 Check the evidence:</strong> What proves the device applied catalog revision 9?</p>
    <label><input type="radio" name="q" value="a"> Backend publish returned without an error</label><br><label><input type="radio" name="q" value="b"> Broker PUBACK</label><br><label><input type="radio" name="q" value="c"> A validated device response/report of applied revision 9</label><br><button type="submit">Check answer</button><p data-feedback aria-live="polite"></p>
  </form>
  <h2>⚙️ Configuration: make races visible</h2>
  <p>A device fetch request should carry a request ID and its known revision. The backend returns the selected version to an authorized response destination; the device validates the request/revision and applies only according to the agreed revision rule. Correlation Data can match a response to a request, but it does not authorize a Response Topic.</p>
  <h2>❤️ Health and logs answer different questions</h2>
  <p>Track broker connection status, heartbeat freshness, and device-reported health as separate observations. Bound log size and frequency, sanitize untrusted content, and avoid dumping raw payloads. Decide which records are disposable telemetry and which deserve durable storage in Lesson 13.</p>
  <h2>🧭 File overview: <code>internal/vending/store.go</code></h2>
  <p>Create this complete file for the Lesson 12 in-memory stage. It owns accepted synthetic sales identities and the latest catalog revision per device. It deliberately does not publish MQTT messages; the existing transport and Section 7 command path remain responsible for wire delivery.</p>
  <h3>💰 <code>RecordSale</code> — short summary</h3><p>Stores a validated report under <code>(device_id, report_id)</code> and returns whether the effect was new, a duplicate, or an ID conflict.</p>
  <h3>📦 <code>ApplyCatalog</code> — short summary</h3><p>Applies only a newer catalog revision and returns a copy of its product list.</p>
  <details class="card" open><summary><strong>🧩 Complete file contents</strong></summary><pre><code>package vending

import (
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

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

type SaleOutcome uint8

const (
	SaleAccepted SaleOutcome = iota + 1
	SaleDuplicate
	SaleIDConflict
)

type CatalogItem struct {
	SKU   string `json:"sku"`
	Price int64  `json:"price_minor"`
}

type Catalog struct {
	SchemaVersion int          `json:"schema_version"`
	JobID         string       `json:"job_id"`
	DeviceID      string       `json:"device_id"`
	Revision      uint64       `json:"revision"`
	Items         []CatalogItem `json:"items"`
}

type Configuration struct {
	SchemaVersion int               `json:"schema_version"`
	RequestID     string            `json:"request_id"`
	DeviceID      string            `json:"device_id"`
	Revision      uint64            `json:"revision"`
	Settings      map[string]string `json:"settings"`
}

type HealthObservation struct {
	DeviceID   string
	Status     string
	ReceivedAt time.Time
}

type Store struct {
	mu       sync.Mutex
	sales    map[string][32]byte
	effects  uint64
	catalogs map[string]Catalog
	configs  map[string]Configuration
	health   map[string]HealthObservation
}

func NewStore() *Store {
	return &amp;Store{
		sales: make(map[string][32]byte), catalogs: make(map[string]Catalog),
		configs: make(map[string]Configuration), health: make(map[string]HealthObservation),
	}
}

// RecordSale uses stable business identity; MQTT packet IDs are not accepted here.
func (s *Store) RecordSale(report contracts.SalesReport) (SaleOutcome, error) {
	if report.DeviceID == "" || report.ReportID == "" {
		return 0, fmt.Errorf("device ID and report ID are required")
	}
	encoded, err := json.Marshal(report)
	if err != nil {
		return 0, fmt.Errorf("encode report identity: %w", err)
	}
	digest := sha256.Sum256(encoded)
	key := report.DeviceID + "/" + report.ReportID
	s.mu.Lock()
	defer s.mu.Unlock()
	if previous, exists := s.sales[key]; exists {
		if previous != digest {
			return SaleIDConflict, nil
		}
		return SaleDuplicate, nil
	}
	s.sales[key] = digest
	s.effects++ // Stand-in for one synthetic business effect in this lesson.
	return SaleAccepted, nil
}

func (s *Store) EffectCount() uint64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.effects
}

// ApplyCatalog replaces device state only for a strictly newer revision.
func (s *Store) ApplyCatalog(catalog Catalog) (bool, error) {
	if catalog.DeviceID == "" || catalog.JobID == "" || catalog.Revision == 0 {
		return false, fmt.Errorf("device ID, job ID, and positive revision are required")
	}
	for _, item := range catalog.Items {
		if item.SKU == "" || item.Price &lt; 0 {
			return false, fmt.Errorf("catalog item has invalid SKU or price")
		}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	previous, exists := s.catalogs[catalog.DeviceID]
	if exists &amp;&amp; catalog.Revision &lt;= previous.Revision {
		return false, nil
	}
	catalog.Items = append([]CatalogItem(nil), catalog.Items...)
	s.catalogs[catalog.DeviceID] = catalog
	return true, nil
}

func (s *Store) CatalogFor(deviceID string) (Catalog, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	catalog, exists := s.catalogs[deviceID]
	if !exists {
		return Catalog{}, false
	}
	catalog.Items = append([]CatalogItem(nil), catalog.Items...)
	return catalog, true
}

// ApplyConfiguration rejects equal or older revisions to make retries idempotent.
func (s *Store) ApplyConfiguration(config Configuration) (bool, error) {
	if config.DeviceID == "" || config.RequestID == "" || config.Revision == 0 || len(config.Settings) == 0 {
		return false, fmt.Errorf("device, request ID, positive revision, and settings are required")
	}
	for key, value := range config.Settings {
		if key == "" || len(key) &gt; 64 || len(value) &gt; 256 || strings.ContainsAny(key, "\r\n") {
			return false, fmt.Errorf("configuration contains an invalid key or oversized value")
		}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	previous, exists := s.configs[config.DeviceID]
	if exists &amp;&amp; config.Revision &lt;= previous.Revision {
		return false, nil
	}
	config.Settings = cloneSettings(config.Settings)
	s.configs[config.DeviceID] = config
	return true, nil
}

func (s *Store) ConfigurationFor(deviceID string) (Configuration, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	config, exists := s.configs[deviceID]
	if !exists { return Configuration{}, false }
	config.Settings = cloneSettings(config.Settings)
	return config, true
}

func (s *Store) RecordHealth(deviceID, status string, receivedAt time.Time) error {
	if deviceID == "" || (status != "ok" &amp;&amp; status != "warning" &amp;&amp; status != "fault") {
		return fmt.Errorf("device ID or health status is invalid")
	}
	s.mu.Lock()
	s.health[deviceID] = HealthObservation{DeviceID: deviceID, Status: status, ReceivedAt: receivedAt.UTC()}
	s.mu.Unlock()
	return nil
}

func (s *Store) HealthFor(deviceID string) (HealthObservation, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	observation, ok := s.health[deviceID]
	return observation, ok
}

// SanitizeLogLine bounds untrusted device log text before it reaches a log sink.
func SanitizeLogLine(message string) string {
	clean := strings.Map(func(r rune) rune {
		if r == '\r' || r == '\n' || r == 0 { return ' ' }
		if r &lt; 0x20 &amp;&amp; r != '\t' { return -1 }
		return r
	}, message)
	clean = strings.TrimSpace(clean)
	if len(clean) &gt; 512 { clean = clean[:512] }
	return clean
}

func cloneSettings(input map[string]string) map[string]string {
	copy := make(map[string]string, len(input))
	for key, value := range input { copy[key] = value }
	return copy
}
</code></pre></details>
  <div class="card"><h3>🔎 Details: what this file proves</h3><p>The SHA-256 digest makes a reused report ID with changed content visible as a conflict. The mutex makes state safe inside this process, but it is not durable across restart and not shared between processes. <code>EffectCount</code> is a lab observation, not a revenue ledger. Catalog slices and configuration maps are copied on write/read so callers cannot mutate stored state through shared backing data. <code>SanitizeLogLine</code> removes line breaks/control bytes and caps log text; it does not identify secrets, so never log arbitrary payloads without an allowlist.</p></div>
  <h2>🧪 Build the workflow matrix</h2>
  <ol><li>Replay one report ID twice and then reuse it with changed content. Define whether changed content is a conflict.</li><li>Apply catalog revision 9, then 8, then a duplicate 9. Record the simulator's reported state and deduplication effect.</li><li>Request config at revision 4 while the backend is preparing revision 5. Define which response is valid.</li><li>Compare a fresh heartbeat with a device health report that says “fault”. Explain why freshness is not health.</li><li>Inspect the existing backend read-only and record its actual contract/identity behavior without assuming lab names match production.</li></ol>
  <div class="card"><h3>🔎 What each result proves</h3><p>Two MQTT deliveries prove the broker delivered twice; one in-memory effect shows this process deduplicated. A backend PUBACK proves its QoS exchange. A device result/report proves application-level evidence it chose to report. None of these in-memory exercises prove crash-safe deduplication; PostgreSQL arrives next.</p></div>
  <h2>✅ Done when</h2><p>Duplicates do not repeat the simulated business effect, stale revisions do not replace newer state, and your workflow notes name the evidence boundary for each operation.</p>
  <details class="card"><summary><strong>📋 Short self-check answers</strong></summary><ol><li>Use stable application report identity, not MQTT packet ID.</li><li>Device-reported applied revision is application evidence; PUBACK is transport evidence.</li><li>Fresh heartbeat, current MQTT connection, and device health measure different facts.</li></ol></details>
  <p>📚 Full workflow exercise and comparison questions: <a href="../course-source/12-vending-workflows.md">Lesson 12 source lesson</a>. Primary reference: <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">OASIS MQTT 5.0 Standard</a> for the transport acknowledgement boundary.</p>
  <p>Ask me to review a duplicate-report or stale-revision result with you before moving on to PostgreSQL.</p>
  <footer><p>Previous: <a href="0011-security-boundaries.html">Lesson 11 — security boundaries</a> · Next: <a href="0013-persist-reports-command-state.html">Lesson 13 — durable persistence</a>.</p></footer>
</main></body></html>