Go MQTT Vending Backend · Lesson 8 · 6 minutes

HTTP acceptance is not device completion

Build an honest response for a vending command that travels from an HTTP caller through your Go service and Mosquitto to a device.

🎯 One skill: name each acknowledgement🧠 Recall before reveal🧪 Disposable broker lab

❓ Start with the system story

A dashboard user asks the backend to ping a vending machine. Your API admitted a command record to process-local memory and published the MQTT request. The broker returned PUBACK, but the machine has not replied yet. What can you tell the caller right now?

Pause and answer aloud before reading on.

📍 Checkout status

internal/httpapi/handlers.go and server.go are Section 8 files to create; they do not yet exist in the current codebase. The complete target files and edited backend entrypoint are included below. They depend on the command service introduced in Section 7.

Lab boundary: the API binds to loopback and has no authentication. Do not expose it to another network; Section 11 adds API authentication and per-device authorization. HTTP 202 admits work into RAM only. A process restart loses command and retry-key state until Sections 13–14 add durable storage and a recoverable outbox.

🗺️ First, see the whole path

The API request, MQTT exchange, and device operation are separate stages. Each can succeed or fail independently.

Caller sends
POST
→
Backend records command
API admission
→
Broker replies
PUBACK
→
Device reports later
application result

HTTP 202 means the server accepted the request for processing; the processing is not complete. A QoS 1 PUBACK is evidence for its MQTT publication exchange. The device's later result is application evidence. The API should return a command ID now and let the caller query status later. Keep the evidence map nearby ↗

🔎 Then, read the function through its purpose

📄 What the file does

The new internal/httpapi/handlers.go will translate HTTP input into command-service calls and command records into HTTP responses. It will not publish MQTT packets itself.

🚀 `createCommand` — short summary

Validates the request and idempotency key, asks the command service to admit one command, then returns its ID and current evidence.

🧩 Details that affect behavior

📦 `writeCommand` — shape the status response

Summary: Sets response metadata, writes the chosen HTTP status, then encodes a command record as JSON.

type commandResponse struct {
    CommandID   string `json:"command_id"`
    DeviceID    string `json:"device_id"`
    Status      string `json:"status"`
    BrokerAcked bool   `json:"broker_acked"`
}

func writeCommand(w http.ResponseWriter, record commands.Record, status int) {
    w.Header().Set("Content-Type", "application/json")
    w.Header().Set("Location", "/commands/"+record.CommandID)
    w.WriteHeader(status)
    _ = json.NewEncoder(w).Encode(commandResponse{
        CommandID: record.CommandID,
        DeviceID: record.DeviceID,
        Status: string(record.Status),
        BrokerAcked: record.BrokerAcked,
    })
}

Details: Location tells the client where to read status. The HTTP status is an argument because the same mapper serves both POST admission and GET status. record.Status and BrokerAcked remain separate fields: the first is application progress, while the second records MQTT publish acknowledgement. The caller should query the returned resource for later device evidence.

This function excerpt focuses on response construction. The full handler file, including imports, request handling, and supporting functions, is included below.

The route patterns shown in this course use Go's method-aware ServeMux syntax and Request.PathValue, available from Go 1.22. This repository's go.mod targets Go 1.27.1. Go team: routing enhancements ↗

🗺️ File and function map

File overview: handlers.go owns HTTP routes, request validation, retry-key handling, and JSON responses. server.go builds a loopback-only server with I/O deadlines. The backend main.go wires HTTP to the command service while continuing to receive MQTT results and expire old commands.

📄 internal/httpapi/handlers.go

Function summaries: NewAPI stores dependencies; Handler registers the two routes; createCommand validates and admits a request; getCommand reads current status; decodeRequest bounds and strictly decodes JSON; writeCommand shapes the JSON response; validPathID checks safe path IDs; newCommandID generates a random stable ID.

🔎 Details

  • createCommand validates device, idempotency key, body size/shape, and allowed command before calling the service. It hashes device plus command as a request fingerprint. Reusing the same key and same input returns the saved record; reusing it with different input returns conflict.
  • The retry map is protected by a mutex, and this simple implementation keeps the lock during service admission. That makes same-key retries atomic in one process, but serializes concurrent creates and loses entries on restart.
  • If the service returns a record plus a publish error, the handler still stores that record and returns 202: the command exists, but its publish outcome may be uncertain. The status body keeps status and broker_acked separate.
  • decodeRequest caps the body, rejects unknown JSON fields, and requires exactly one JSON value. writeCommand sets Location before writing the status and body; after WriteHeader, headers cannot be changed.

🌐 internal/httpapi/server.go

NewServer summary: accepts only a loopback address with an explicit port and attaches the API handler plus finite read/write/idle timeouts.

🔎 Details

Loopback keeps this teaching endpoint local; it is not an internet-facing production server. Header, body, response, and idle timeouts limit resource use from slow clients. Production exposure would need an intentional network binding, authentication/authorization, TLS termination, and operational configuration.

🔌 cmd/backend/main.go

run summary: starts MQTT, wires store → command service → HTTP API, runs HTTP in a goroutine, processes MQTT results and expiry ticks, then shuts down HTTP on cancellation.

main summary: reports a startup/runtime error and exits nonzero.

🔎 Details

MQTT starts before the API is exposed, so a request cannot be accepted when the command publisher has no initialized connection manager. The main select loop continues processing response messages while HTTP serves requests; otherwise commands could never advance to device-reported results. The ticker changes pending/accepted commands to unknown after deadlines. On a signal, Server.Shutdown stops accepting new HTTP requests and waits up to five seconds for in-flight handlers. The in-memory store still loses all status on process restart.

🧪 Check your model

Choose the evidence each event actually provides. Options in each question have the same word count; the correct answer moves between positions.

1. What does HTTP 202 support?
2. What does a successful QoS 1 PUBACK establish?
3. What can GET /commands/{id} return later?

🛠️ Your tiny design task

Open the Section 8 implementation and find writeCommand. Without changing code yet, write the response a caller should see immediately after command admission:

  1. Which HTTP status should it return?
  2. Which stable identifier should the caller save?
  3. What URL should the Location header name?
  4. Which status field must not be described as “completed” just because PUBACK arrived?

Then compare your answer to the existing implementation in Section 8 — Expose device commands through HTTP ↗. If it matches, explain why in your own words. If it does not, identify which component owns the evidence.

📦 Section 8 complete HTTP 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/httpapi/handlers.go

package httpapi

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"

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

const maxRequestBytes = 4 * 1024

// The API depends on command operations, not on the MQTT client itself.
type commandService interface {
	CreateAndPublish(ctx context.Context, deviceID, commandID string) (commands.Record, error)
	Get(commandID string) (commands.Record, bool)
}

type retryEntry struct {
	fingerprint [32]byte
	record      commands.Record
}

type API struct {
	commands   commandService
	serviceCtx context.Context
	mu         sync.Mutex
	retries    map[string]retryEntry
}

// NewAPI keeps retries in memory and ties accepted work to the app lifecycle.
func NewAPI(serviceCtx context.Context, service commandService) *API {
	return &API{commands: service, serviceCtx: serviceCtx, retries: make(map[string]retryEntry)}
}

type createRequest struct {
	Command string `json:"command"`
}
type commandResponse struct {
	CommandID   string `json:"command_id"`
	DeviceID    string `json:"device_id"`
	Status      string `json:"status"`
	BrokerAcked bool   `json:"broker_acked"`
	Detail      string `json:"detail,omitempty"`
}

func (a *API) Handler() http.Handler {
	// Keep the public surface small: create commands and read their status.
	mux := http.NewServeMux()
	mux.HandleFunc("POST /devices/{deviceID}/commands", a.createCommand)
	mux.HandleFunc("GET /commands/{commandID}", a.getCommand)
	return mux
}

func (a *API) createCommand(w http.ResponseWriter, r *http.Request) {
	// Validate caller input before reserving an idempotency key or publishing.
	deviceID := r.PathValue("deviceID")
	if !validPathID(deviceID) {
		http.Error(w, "invalid device ID", http.StatusBadRequest)
		return
	}
	key := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
	if key == "" || len(key) > 128 {
		http.Error(w, "valid Idempotency-Key required", http.StatusBadRequest)
		return
	}
	var input createRequest
	if err := decodeRequest(w, r, &input); err != nil {
		var tooLarge *http.MaxBytesError
		if errors.As(err, &tooLarge) {
			http.Error(w, "request body too large", http.StatusRequestEntityTooLarge)
			return
		}
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if input.Command != "ping" {
		http.Error(w, "unsupported command", http.StatusBadRequest)
		return
	}
	// The fingerprint detects reuse of one key for different request content.
	fingerprint := sha256.Sum256([]byte(deviceID + "\x00" + input.Command))

	// This in-memory lesson serializes admissions to make same-key retries atomic.
	a.mu.Lock()
	defer a.mu.Unlock()
	if previous, ok := a.retries[key]; ok {
		if previous.fingerprint != fingerprint {
			http.Error(w, "idempotency key reused with different request", http.StatusConflict)
			return
		}
		writeCommand(w, previous.record, http.StatusAccepted)
		return
	}
	commandID, err := newCommandID()
	if err != nil {
		http.Error(w, "could not create command ID", http.StatusInternalServerError)
		return
	}
	record, publishErr := a.commands.CreateAndPublish(a.serviceCtx, deviceID, commandID)
	if record.CommandID == "" {
		if publishErr != nil {
			http.Error(w, "command admission unavailable", http.StatusServiceUnavailable)
			return
		}
		http.Error(w, "command admission failed", http.StatusServiceUnavailable)
		return
	}
	a.retries[key] = retryEntry{fingerprint: fingerprint, record: record}
	// The command record exists even if the publish outcome is uncertain.
	writeCommand(w, record, http.StatusAccepted)
}

func (a *API) getCommand(w http.ResponseWriter, r *http.Request) {
	record, ok := a.commands.Get(r.PathValue("commandID"))
	if !ok {
		http.Error(w, "command not found", http.StatusNotFound)
		return
	}
	writeCommand(w, record, http.StatusOK)
}

// decodeRequest enforces a byte limit, strict fields, and exactly one JSON value.
func decodeRequest(w http.ResponseWriter, r *http.Request, target any) error {
	r.Body = http.MaxBytesReader(w, r.Body, maxRequestBytes)
	decoder := json.NewDecoder(r.Body)
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(target); err != nil {
		return fmt.Errorf("invalid JSON request: %w", err)
	}
	var trailing any
	if err := decoder.Decode(&trailing); err != io.EOF {
		if err == nil {
			return errors.New("request must contain one JSON value")
		}
		return fmt.Errorf("trailing request data: %w", err)
	}
	return nil
}

func writeCommand(w http.ResponseWriter, record commands.Record, status int) {
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Location", "/commands/"+record.CommandID)
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(commandResponse{CommandID: record.CommandID, DeviceID: record.DeviceID, Status: string(record.Status), BrokerAcked: record.BrokerAcked, Detail: record.Detail})
}

// Path IDs use the same restricted alphabet as contract and topic IDs.
func validPathID(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
}

func newCommandID() (string, error) {
	var raw [16]byte
	if _, err := rand.Read(raw[:]); err != nil {
		return "", err
	}
	return "cmd-" + hex.EncodeToString(raw[:]), nil
}

📄 internal/httpapi/server.go

package httpapi

import (
	"fmt"
	"net"
	"net/http"
	"time"
)

func NewServer(address string, api *API) (*http.Server, error) {
	// The teaching API listens only on loopback and uses finite I/O timeouts.
	host, port, err := net.SplitHostPort(address)
	if err != nil || (host != "127.0.0.1" && host != "localhost") || port == "" {
		return nil, fmt.Errorf("HTTP lab address must be loopback with an explicit port")
	}
	return &http.Server{
		Addr: address, Handler: api.Handler(),
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       10 * time.Second,
		WriteTimeout:      10 * time.Second,
		IdleTimeout:       60 * time.Second,
	}, nil
}

📄 cmd/backend/main.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/commands"
	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/httpapi"
	"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()
	inbox := make(chan mqtt.Message, 16)
	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 }
	topics := []contracts.TopicFilter{telemetryFilter, responsesFilter, salesFilter}
	manager, err := mqtt.Start(ctx, cfg, topics, inbox)
	if err != nil {
		return fmt.Errorf("start MQTT: %w", err)
	}
	defer mqtt.Stop(manager)

	// Wire the command service before exposing its HTTP endpoints.
	store := commands.NewStore()
	publisher := mqtt.CommandPublisher{Manager: manager}
	service := commands.NewService(store, publisher, 15*time.Second)
	api := httpapi.NewAPI(ctx, service)
	server, err := httpapi.NewServer("127.0.0.1:18080", api)
	if err != nil {
		return err
	}
	serverErrors := make(chan error, 1)
	go func() { serverErrors <- server.ListenAndServe() }()
	log.Printf("HTTP API listening on %s", server.Addr)
	expirationTicker := time.NewTicker(time.Second)
	defer expirationTicker.Stop()

	// Keep response processing and expiration active for the server lifetime.
	for {
		select {
		case <-ctx.Done():
			shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if err := server.Shutdown(shutdown); err != nil {
				return fmt.Errorf("HTTP shutdown: %w", err)
			}
			return nil
		case err := <-serverErrors:
			if !errors.Is(err, http.ErrServerClosed) {
				return fmt.Errorf("HTTP server: %w", err)
			}
			return nil
		case <-expirationTicker.C:
			service.Expire(time.Now().UTC())
		case message := <-inbox:
			topic, err := contracts.ParseTopic(message.Topic)
			if err != nil {
				log.Printf("rejected topic=%q reason=%v", message.Topic, err)
				continue
			}
			switch topic.Kind {
			case contracts.KindResponses:
				if err := commands.HandleMessage(service, message); err != nil {
					log.Printf("rejected response topic=%q reason=%v", message.Topic, err)
				}
			case contracts.KindTelemetry:
				var heartbeat contracts.Heartbeat
				heartbeat, err = contracts.Decode[contracts.Heartbeat](message.Payload, topic, time.Now().UTC())
				if err != nil {
					log.Printf("rejected telemetry topic=%q reason=%v", message.Topic, err)
					continue
				}
				log.Printf("valid heartbeat device=%s message=%s", heartbeat.DeviceID, heartbeat.MessageID)
			case contracts.KindSales:
				var report contracts.SalesReport
				report, err = contracts.Decode[contracts.SalesReport](message.Payload, topic, time.Now().UTC())
				if err != nil {
					log.Printf("rejected sales topic=%q reason=%v", message.Topic, err)
					continue
				}
				log.Printf("valid sales report device=%s report=%s", report.DeviceID, report.ReportID)
			default:
				log.Printf("ignored topic kind=%s", topic.Kind)
			}
		}
	}
}

🏗️ Build and run the HTTP-to-device path

After saving the Section 7 command files plus the Section 8 HTTP files and backend wiring, build and start the lab 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

The tests cover only the tests present in the checkout; the builds verify the complete API and simulator programs compile. 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

Once the backend reports that the HTTP API is listening, submit a harmless ping. Copy the returned command_id into the GET request:

curl -i -X POST http://127.0.0.1:18080/devices/device-001/commands \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: lesson8-ping-001' \
  --data '{"command":"ping"}'

# Replace cmd-example with the command_id returned by POST.
curl -i http://127.0.0.1:18080/commands/cmd-example

Expect HTTP 202 from POST, then later application status from GET after the simulator responds. Repeat the same POST with the same idempotency key and body; it should return the same command ID. PUBACK alone is not device completion. Stop both clients with Ctrl+C, then stop the broker with docker compose -p mqtt5-learning stop broker. Ask me a follow-up if the observed status or retry result differs from your prediction.

✅ Takeaway

202 = API admission. PUBACK = MQTT publication acknowledgement. Device result = application-reported outcome.

Keep those facts separate in logs, API fields, dashboards, and incident reports. Ask me a follow-up question if any boundary still feels fuzzy.