mqtt-5-learning/teaching/course-source/08-http-commands.md

# 🌐 Section 8 — Expose device commands through HTTP

## 🎬 Start with the system story

> **❓ Question:** **A user clicks “ping device” in an HTTP API. How can the API respond promptly while the MQTT command and device response happen asynchronously?**

The HTTP handler validates the request and asks an application service to create a command. The service records its identity and pending status, publishes through the MQTT client, then exposes later device evidence through a status endpoint. HTTP request lifetime, MQTT delivery, and the device operation each have their own acknowledgement and failure boundary.

```mermaid
sequenceDiagram
    participant U as API caller
    participant H as Go HTTP handler
    participant S as Command service
    participant B as MQTT broker
    participant D as Device
    U->>H: POST /devices/{id}/commands
    H->>S: Validate and create command
    S->>S: Store command ID and pending state
    S->>B: Publish MQTT command
    B-->>S: PUBACK for broker hop
    S-->>H: Accepted with command ID
    H-->>U: HTTP 202 + command ID
    B->>D: Deliver command
    D->>B: Publish application result
    B->>S: Deliver result
    U->>H: GET /commands/{id}
    H->>S: Read current status
    S-->>H: Accepted/completed/rejected/unknown
    H-->>U: HTTP status representation
```

**Before each implementation step, ask:** Is this response evidence that the HTTP request was accepted, that the broker acknowledged a publication, or that the device completed the work? Follow [Lesson 0](00-system-lifecycle.md) for the complete MQTT lifecycle.

Prerequisite: Section 7 provides a command service and queryable in-memory states. The backend and simulator continue to read their shared disposable password from `configs/lab.toml`; select each broker username with `-username` when starting the process.

**Scope and trust boundary:** This lesson's HTTP endpoint is unauthenticated and binds to loopback for the disposable lab. Do not expose it to another network; Section 11 adds API authentication and per-device authorization. HTTP 202 means the command was admitted to process-local memory, not durably stored. A restart loses the command and retry-key maps until Sections 13–14 add persistence and recoverable publication intent.

Run all commands from `/home/ryan/Media/Code/Code/robot-market/mqtt-5-learning`. Save the complete HTTP implementation blocks at the project-relative paths shown below. This lesson includes the handlers and server setup; the remaining exercises trace request, retry, and device-result behavior.

## 🌐 Translate evidence between HTTP and MQTT

HTTP has one response to this request. MQTT has an asynchronous stream where the device may reply after the HTTP connection closes. Returning 202 after admission gives the caller a stable command ID; a later GET reports evidence accumulated by the command service.

```text
POST + retry key → admit command C → HTTP 202 + C
                                   → MQTT publication → device result → status C
GET C --------------------------------------------------------------→ status C
```

**❓ Question:** **The backend publishes a ping and receives PUBACK. Does that mean the device accepted or completed it?**

**💬 Answer:** No. PUBACK confirms the broker acknowledged the backend's publication. It does not confirm device delivery or business acceptance. The HTTP API returns a command ID, and a later status request reports the device's application result if one arrives. The handler uses the request context while parsing. After validation, it passes work to the command service with the application lifecycle context and the command's own deadline. A disconnected caller therefore does not cancel accepted work.

Predict a lost HTTP response: the server admits C, but the caller never sees 202 and retries. The same retry key and payload should return C, not create D. Reusing the key with a changed payload is a conflict. Later, scope keys by authenticated caller so two users cannot accidentally share a command record.

<details>
<summary>If the backend restarts after returning 202, can this lesson recover C?</summary>

No. The lesson's in-memory command and retry-key maps disappear. HTTP 202 describes admission under the current implementation, not disk durability. Sections 13–14 add database admission and the outbox; Section 15 separates the processes. Until then, document this restart limit in the lab API.

</details>

## ⚙️ 1. Add HTTP configuration and handlers

> **❓ Question:** **Which process accepts the user's HTTP request, and how does that process reach the command service without putting MQTT logic in the handler?**

Wire the HTTP server to the application service and preserve bounded parsing, deadlines, and controlled shutdown.

Save the complete API implementation below as `internal/httpapi/handlers.go` and `server.go`. It binds only to loopback, limits and strictly decodes request bodies, requires an idempotency key, and sets finite HTTP timeouts. The lab listener address is fixed in the complete backend entrypoint below; production authorization is introduced later.

`internal/httpapi/handlers.go`:

```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`:

```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
}
```

The complete final `cmd/backend/main.go` is below. It starts MQTT before HTTP, routes response messages into the command service, and shuts down the HTTP server before disconnecting MQTT. It replaces the Section 5 backend entrypoint.

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

## 📋 2. Define the API contract

> **❓ Question:** **What should the caller receive immediately when device work will finish later?**

Return an accepted command ID, then provide a status resource that reports later evidence.

The complete handlers above implement `POST /devices/{id}/commands` and `GET /commands/{id}`. They limit bodies, reject trailing JSON and unknown fields, validate the path ID, require an idempotency key, and return consistent JSON status. The listener remains local; production user authorization is introduced later.

Return HTTP 202 with a command ID and a Location pointing to its status after service admission. Document malformed-input 400, oversized-body 413, unknown-command/device 404 or a consistent chosen alternative, conflict 409, and admission-unavailable 503 behavior. Transport uncertainty after acceptance belongs in command status rather than silently discarding an accepted command.

## 🌐 3. Implement HTTP retry identity

> **❓ Question:** **If the server accepted a request but the HTTP response was lost, how can the caller retry without creating a second device action?**

Use an HTTP idempotency key with stable request semantics; keep it distinct from the command ID and MQTT packet identifier.

The handler keeps the key, request fingerprint, and command record together in memory. An identical retry returns the original ID; changed content under that key returns 409. The mutex makes admission atomic in this single-process lesson. A restart loses this map, so this is not a durable idempotency guarantee. The HTTP key, command ID, and MQTT packet ID have different lifetimes and purposes.

Use the HTTP context for request parsing and admission. Once accepted, execute under the application's lifecycle and command deadline, so client disconnect does not cancel the command. Inject the command service through a small interface and keep MQTT calls out of handlers.

## 🧪 4. Run and verify

> **❓ Question:** **What evidence demonstrates each boundary in the round trip—from HTTP acceptance to broker delivery to device result?**

Observe the returned command ID, MQTT acknowledgement, simulator response, and status endpoint separately.

Start the simulator and backend in separate terminals from the project root. The test password is already loaded from `configs/lab.toml`; `-username` selects the account for each process:

```sh
go run ./cmd/simulator -username simulator -client-id lab-simulator-device-001
```

```sh
go run ./cmd/backend -username backend -client-id lab-backend
```

In a third terminal, send the HTTP command:

```sh
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 ID returned by the POST response.
curl -i http://127.0.0.1:18080/commands/cmd-example
```

`curl` is an HTTP client. `-i` prints response headers alongside the body; `-X POST` chooses the method; each `-H` sets a request header; `--data` sends the quoted JSON bytes. Content-Type tells the server how to interpret the body, while Idempotency-Key identifies this caller retry. This local exercise sends no production authorization token.

Copy the returned ID into `GET /commands/{id}`. Repeat the identical POST and confirm the same ID. Use a new key for a genuinely new command.

For example, if the returned ID is `cmd-example`, use `curl -i http://127.0.0.1:18080/commands/cmd-example`; without `--data` or an explicit method, curl performs GET. Replace that example ID with the actual returned value. HTTP 202 means admitted for asynchronous processing; inspect later status for device completion. `httptest` runs handlers against an in-process request/recorder or test server, letting you assert headers, status, and body without the live broker.

Set ReadHeaderTimeout, ReadTimeout, WriteTimeout, and IdleTimeout deliberately. Coordinate bounded HTTP shutdown with the MQTT and worker lifecycle.

```sh
mkdir -p internal/httpapi

gofmt -w internal cmd

go test -race ./...

go vet ./...
```

ReadHeaderTimeout bounds header reading; ReadTimeout bounds reading the request including its body; WriteTimeout bounds response writing; IdleTimeout bounds waiting for another request on an idle keep-alive connection. Choose values that fit this asynchronous API rather than waiting in POST for a device result. Server `Shutdown` stops accepting work and waits for active HTTP requests within its context deadline; your application must separately drain accepted command work.

`mkdir -p` creates the HTTP package directory, `gofmt` formats edits, and the test/race/vet commands check behavior, executed concurrent access, and suspicious static patterns respectively. Run directory creation before writing the files even though the check block collects the commands together.

The complete handler performs the same checks an `httptest` would exercise: strict decoding, body limits, idempotency, unavailable admission, unknown IDs, and status responses. Run the supplied local `curl` flow to observe them through the real HTTP server. Accepted work is owned by the command service; device completion arrives later through MQTT.

**Done when:** HTTP creates a ping, status reflects device evidence, retries reuse its ID, and closing the caller connection does not erase accepted work.

## ✅ Result — the story so far

An API caller can now ask the backend to ping a device with HTTP. The handler validates the request and admits it through the command service; the caller receives HTTP 202 and a command ID without waiting for the device. The backend publishes the MQTT command, the simulator returns its application result, and a later GET reports the command's current evidence. Retrying the same POST with the same idempotency key returns the original command instead of starting a second effect. HTTP acceptance, MQTT acknowledgement, and device completion remain separate events.

Previous: [07-command-response.md](07-command-response.md). Next: [09-sessions-recovery.md](09-sessions-recovery.md).