Go MQTT Vending Backend · Lesson 5A · 15 minutes

Add the backend message-processing step

Section 5 validates incoming bytes. In 5A, add the application code that turns a valid heartbeat into a small, queryable backend state.

🗺️ First, see what the files do

Validation is a trust boundary, not a business action. The service applies the meaning of a valid message. The store owns the resulting view. Keeping these roles separate lets Section 6 move service work to workers without mixing application state with Paho callback code.

🗃️ Then, create `internal/deviceview/store.go`

File overview: this file owns one latest heartbeat snapshot per device. It keeps the map in process memory, so a backend restart clears the view.

`NewStore` summary: initializes the empty map. A query before the first heartbeat can report “not seen yet” rather than inventing an offline record.

`RecordHeartbeat` summary: parses device-reported time, copies optional temperature, and replaces the snapshot only when the report is newer.

`GetStatus` summary: returns the snapshot, plus whether the backend received it within the requested freshness window.

🧩 Complete file

package deviceview

import (
	"fmt"
	"sync"
	"time"

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

type Snapshot struct {
	DeviceID    string
	MessageID   string
	ReportedAt  time.Time
	ReceivedAt  time.Time
	Temperature    float64
	TemperatureSet bool
}

type Status struct {
	Snapshot       Snapshot
	HeartbeatFresh bool
}

type Store struct {
	mu     sync.RWMutex
	latest map[string]Snapshot
}

func NewStore() *Store {
	return &Store{latest: make(map[string]Snapshot)}
}

func (s *Store) RecordHeartbeat(message contracts.Heartbeat, receivedAt time.Time) (bool, error) {
	reportedAt, err := time.Parse(time.RFC3339, message.OccurredAt)
	if err != nil {
		// Decode already validates this timestamp. Keep the store defensive too.
		return false, fmt.Errorf("parse heartbeat timestamp: %w", err)
	}

	snapshot := Snapshot{
		DeviceID:   message.DeviceID,
		MessageID:  message.MessageID,
		ReportedAt: reportedAt,
		ReceivedAt: receivedAt,
	}
	if message.Temperature != nil {
		snapshot.Temperature = *message.Temperature
		snapshot.TemperatureSet = true
	}

	s.mu.Lock()
	defer s.mu.Unlock()
	previous, exists := s.latest[message.DeviceID]
	if exists && !reportedAt.After(previous.ReportedAt) {
		return false, nil
	}
	s.latest[message.DeviceID] = snapshot
	return true, nil
}

func (s *Store) GetStatus(deviceID string, now time.Time, freshFor time.Duration) (Status, bool) {
	s.mu.RLock()
	snapshot, exists := s.latest[deviceID]
	s.mu.RUnlock()
	if !exists {
		return Status{}, false
	}

	fresh := freshFor > 0 && !now.Before(snapshot.ReceivedAt) && now.Sub(snapshot.ReceivedAt) <= freshFor
	return Status{Snapshot: snapshot, HeartbeatFresh: fresh}, true
}

🧭 Next, create `internal/deviceview/service.go`

File overview: this file receives the existing transport-neutral mqtt.Message, validates it with contracts, applies the behavior assigned to its message kind, and logs a safe outcome. It does not own a Paho client or publish MQTT messages.

`NewService` summary: checks that the store, logger, and positive freshness window are configured before message handling begins.

`HandleMessage` summary: checks cancellation, parses the route, captures one backend receive time, then validates and handles the message according to its kind.

`Status` summary: asks the store for the latest view using the service clock and configured freshness window.

🧩 Complete file

package deviceview

import (
	"context"
	"fmt"
	"log/slog"
	"time"

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

type Service struct {
	store    *Store
	logger   *slog.Logger
	freshFor time.Duration
	now      func() time.Time
}

func NewService(store *Store, logger *slog.Logger, freshFor time.Duration) (*Service, error) {
	if store == nil {
		return nil, fmt.Errorf("device-view store is required")
	}
	if logger == nil {
		return nil, fmt.Errorf("logger is required")
	}
	if freshFor <= 0 {
		return nil, fmt.Errorf("heartbeat freshness window must be positive")
	}
	return &Service{store: store, logger: logger, freshFor: freshFor, now: time.Now}, nil
}

func (s *Service) HandleMessage(ctx context.Context, message mqtt.Message) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return fmt.Errorf("parse topic: %w", err)
	}
	receivedAt := s.now().UTC()
	switch topic.Kind {
	case contracts.KindTelemetry:
		heartbeat, err := contracts.Decode[contracts.Heartbeat](message.Payload, topic, receivedAt)
		if err != nil {
			return fmt.Errorf("validate heartbeat: %w", err)
		}
		updated, err := s.store.RecordHeartbeat(heartbeat, receivedAt)
		if err != nil {
			return fmt.Errorf("record heartbeat: %w", err)
		}
		if !updated {
			s.logger.Info("older or duplicate heartbeat ignored",
				"device_id", heartbeat.DeviceID,
				"message_id", heartbeat.MessageID,
			)
			return nil
		}
		s.logger.Info("heartbeat recorded",
			"device_id", heartbeat.DeviceID,
			"message_id", heartbeat.MessageID,
			"received_at", receivedAt,
		)
		return nil
	case contracts.KindSales:
		report, err := contracts.Decode[contracts.SalesReport](message.Payload, topic, receivedAt)
		if err != nil {
			return fmt.Errorf("validate sales report: %w", err)
		}
		// Validation is not durable admission; Section 13 stores reports.
		s.logger.Info("sales report validated; persistence not implemented",
			"device_id", report.DeviceID,
			"report_id", report.ReportID,
			"item_count", len(report.Items),
		)
		return nil
	case contracts.KindResponses:
		result, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, receivedAt)
		if err != nil {
			return fmt.Errorf("validate command result: %w", err)
		}
		// Section 7 adds pending-command and correlation matching.
		s.logger.Info("command result validated; matching not implemented",
			"device_id", result.DeviceID,
			"command_id", result.CommandID,
			"status", result.Status,
		)
		return nil
	default:
		return fmt.Errorf("device-view service does not handle topic kind %q", topic.Kind)
	}
}

func (s *Service) Status(deviceID string) (Status, bool) {
	return s.store.GetStatus(deviceID, s.now().UTC(), s.freshFor)
}

🔌 Finally, edit `cmd/backend/main.go`

File overview: this existing file loads configuration, starts the MQTT client, owns the inbox receive loop, and periodically publishes the lab's ping command. The edited version keeps those responsibilities and delegates inbound messages to the new service instead of printing raw payloads.

run summary: constructs one store and service before MQTT startup, then passes each received message to handleMessage.

handleMessage summary: keeps the select loop focused on receiving events by delegating application behavior to Service.

🧩 Complete edited file

package main

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

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/deviceview"
	"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("failed to load config: %w", err)
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// Keep one shared in-memory view for the lifetime of this backend process.
	store := deviceview.NewStore()
	deviceView, err := deviceview.NewService(store, slog.Default(), 90*time.Second)
	if err != nil {
		return fmt.Errorf("create device-view service: %w", err)
	}

	telemetryFilter, err := contracts.DeviceFilter(contracts.KindTelemetry)
	if err != nil { return err }
	responsesFilter, err := contracts.DeviceFilter(contracts.KindResponses)
	if err != nil { return err }
	commandTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindCommands)
	if err != nil { return err }
	inbox := make(chan mqtt.Message, 16)

	manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{telemetryFilter, responsesFilter}, inbox)
	if err != nil {
		return fmt.Errorf("failed to start MQTT client: %w", err)
	}
	defer mqtt.Stop(manager)

	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	runID := time.Now().UnixNano()
	sequence := 0

	for {
		select {
		case <-ctx.Done():
			log.Print("shutting down")
			return nil
		case message := <-inbox:
			if err := handleMessage(ctx, deviceView, message); err != nil {
				slog.Error("message handling failed", "topic", message.Topic, "error", err)
			}
		case <-ticker.C:
			sequence++
			id := fmt.Sprintf("ping-%d-%d", runID, sequence)
			payload, err := json.Marshal(contracts.Command{
				SchemaVersion: contracts.SchemaVersion,
				MessageID: id,
				DeviceID: cfg.MQTT.DeviceID,
				OccurredAt: time.Now().UTC().Format(time.RFC3339),
				CommandID: id,
				Command: "ping",
			})
			if err != nil {
				return fmt.Errorf("failed to marshal payload: %w", err)
			}
			if err := mqtt.Publish(ctx, manager, commandTopic, payload); err != nil {
				log.Printf("command publication failed or outcome uncertain: %v", err)
			}
		}
	}
}

func handleMessage(ctx context.Context, deviceView *deviceview.Service, message mqtt.Message) error {
	return deviceView.HandleMessage(ctx, message)
}

🧪 Check your model

🛠️ Your tiny practice

🏗️ Build and run the vertical slice

After saving the new internal/deviceview files and updating cmd/backend/main.go, compile the real programs and start the isolated broker. Run these from the repository root:

go test ./...
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

go test ./... runs the tests that exist; it does not prove every device-view branch is covered. The two builds catch compile and link errors in the actual processes. Check that the project-named broker is running before connecting clients.

Start each binary in its own terminal:

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

/tmp/mqtt5-lab-backend -username backend -client-id lab-backend

Within a few seconds, expect backend log entries for recorded heartbeats with device and message IDs. The backend also sends a ping every ten seconds; the simulator replies, and the backend validates that response. This confirms a real MQTT message reached the application service. It does not establish command correlation, which Lesson 7 adds.

Stop both clients with Ctrl+C, then run docker compose -p mqtt5-learning stop broker. The device view is in memory and disappears when the backend exits. The lesson's temporary deviceView.Status diagnostic can show the snapshot and freshness flag; no HTTP status endpoint exists yet. Ask me a follow-up if a build error or log result is unclear.