Match command results to the right request

📄 internal/commands/store.go

package commands

import (
	"bytes"
	"fmt"
	"sync"
	"time"

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

type Status string

// Status records application progress independently from BrokerAcked.
const (
	Pending   Status = "pending"
	Accepted  Status = "accepted"
	Completed Status = "completed"
	Rejected  Status = "rejected"
	Unknown   Status = "unknown"
)

type Record struct {
	CommandID       string
	DeviceID        string
	CorrelationData []byte
	CreatedAt       time.Time
	Deadline        time.Time
	Status          Status
	BrokerAcked     bool
	AckReasonCode   byte
	Detail          string
}

type Store struct {
	mu      sync.RWMutex
	records map[string]Record
}

// NewStore creates the in-memory command record table.
func NewStore() *Store { return &Store{records: make(map[string]Record)} }

func key(deviceID, commandID string) string { return deviceID + "/" + commandID }

func (s *Store) Insert(record Record) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := key(record.DeviceID, record.CommandID)
	if _, exists := s.records[k]; exists {
		return fmt.Errorf("command already exists")
	}
	record.CorrelationData = append([]byte(nil), record.CorrelationData...)
	s.records[k] = record
	return nil
}

// Get and Update copy correlation bytes so callers cannot mutate stored state.
func (s *Store) Get(deviceID, commandID string) (Record, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	record, ok := s.records[key(deviceID, commandID)]
	record.CorrelationData = append([]byte(nil), record.CorrelationData...)
	return record, ok
}

func (s *Store) Update(record Record) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := key(record.DeviceID, record.CommandID)
	if _, exists := s.records[k]; !exists {
		return fmt.Errorf("command not found")
	}
	record.CorrelationData = append([]byte(nil), record.CorrelationData...)
	s.records[k] = record
	return nil
}

func (s *Store) MarkBrokerAck(deviceID, commandID string, reason byte) (Record, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := key(deviceID, commandID)
	record, ok := s.records[k]
	if !ok {
		return Record{}, fmt.Errorf("command not found")
	}
	record.BrokerAcked, record.AckReasonCode = true, reason
	s.records[k] = record
	return record, nil
}

// MarkUnknownIfPending preserves any application result that arrived first.
func (s *Store) MarkUnknownIfPending(deviceID, commandID, detail string) (Record, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := key(deviceID, commandID)
	record, ok := s.records[k]
	if !ok {
		return Record{}, fmt.Errorf("command not found")
	}
	if record.Status == Pending || record.Status == Accepted {
		record.Status, record.Detail = Unknown, detail
	}
	s.records[k] = record
	return record, nil
}

// ApplyResult checks both device scope and correlation before changing status.
func (s *Store) ApplyResult(deviceID string, correlation []byte, result contracts.CommandResult) (Record, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := key(deviceID, result.CommandID)
	record, ok := s.records[k]
	if !ok {
		return Record{}, fmt.Errorf("no pending command matches result")
	}
	if result.DeviceID != record.DeviceID || !bytes.Equal(correlation, record.CorrelationData) {
		return Record{}, fmt.Errorf("result identity or correlation does not match pending command")
	}
	if record.Status == Completed || record.Status == Rejected {
		return record, nil
	}
	switch result.Status {
	case "accepted":
		record.Status = Accepted
	case "completed":
		record.Status = Completed
	case "rejected":
		record.Status = Rejected
	default:
		return Record{}, fmt.Errorf("unsupported result status %q", result.Status)
	}
	record.Detail = result.Detail
	s.records[k] = record
	return record, nil
}

func (s *Store) List() []Record {
	s.mu.RLock()
	defer s.mu.RUnlock()
	result := make([]Record, 0, len(s.records))
	for _, record := range s.records {
		record.CorrelationData = append([]byte(nil), record.CorrelationData...)
		result = append(result, record)
	}
	return result
}

📄 internal/commands/service.go

package commands

import (
	"context"
	"crypto/rand"
	"fmt"
	"time"

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

type Publisher interface {
	PublishCommand(context.Context, string, contracts.Command, contracts.TopicName, []byte) (byte, error)
}

// Service coordinates command identity, storage, publication, and deadlines.
type Service struct {
	store     *Store
	publisher Publisher
	now       func() time.Time
	timeout   time.Duration
}

func NewService(store *Store, publisher Publisher, timeout time.Duration) *Service {
	return &Service{store: store, publisher: publisher, now: func() time.Time { return time.Now().UTC() }, timeout: timeout}
}

func (s *Service) CreateAndPublish(ctx context.Context, deviceID, commandID string) (Record, error) {
	responseTopic, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return Record{}, fmt.Errorf("build response topic: %w", err) }
	// Store pending state first so a fast device response can match immediately.
	now := s.now()
	correlation := make([]byte, 16)
	if _, err := rand.Read(correlation); err != nil {
		return Record{}, fmt.Errorf("create correlation data: %w", err)
	}
	record := Record{CommandID: commandID, DeviceID: deviceID, CorrelationData: correlation, CreatedAt: now, Deadline: now.Add(s.timeout), Status: Pending}
	if err := s.store.Insert(record); err != nil {
		return Record{}, err
	}
	command := contracts.Command{SchemaVersion: contracts.SchemaVersion, MessageID: "message-" + commandID, DeviceID: deviceID, OccurredAt: now.Format(time.RFC3339), CommandID: commandID, Command: "ping"}
	ctx, cancel := context.WithDeadline(ctx, record.Deadline)
	defer cancel()
	reason, err := s.publisher.PublishCommand(ctx, deviceID, command, responseTopic, append([]byte(nil), correlation...))
	if err != nil {
		updated, _ := s.store.MarkUnknownIfPending(deviceID, commandID, "publish outcome uncertain: "+err.Error())
		return updated, err
	}
	updated, err := s.store.MarkBrokerAck(deviceID, commandID, reason)
	return updated, err
}

func (s *Service) HandleResult(deviceID string, correlation []byte, result contracts.CommandResult) (Record, error) {
	return s.store.ApplyResult(deviceID, correlation, result)
}

// Expire records uncertainty; it does not claim that the device did no work.
func (s *Service) Expire(now time.Time) {
	for _, record := range s.store.List() {
		if (record.Status == Pending || record.Status == Accepted) && !now.Before(record.Deadline) {
			_, _ = s.store.MarkUnknownIfPending(record.DeviceID, record.CommandID, "response deadline expired; execution outcome is unknown")
		}
	}
}

func (s *Service) Get(commandID string) (Record, bool) {
	for _, record := range s.store.List() {
		if record.CommandID == commandID {
			return record, true
		}
	}
	return Record{}, false
}

📄 internal/mqtt/command_publisher.go

package mqtt

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"example.com/mqtt5-learning/internal/contracts"
	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

type CommandPublisher struct {
	Manager *autopaho.ConnectionManager
}

func (p CommandPublisher) PublishCommand(ctx context.Context, deviceID string, command contracts.Command, responseTopic contracts.TopicName, correlation []byte) (byte, error) {
	// Authorize the reply destination before putting it in the MQTT packet.
	authorized, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	if err != nil { return 0, fmt.Errorf("build response topic: %w", err) }
	if responseTopic != authorized {
		return 0, fmt.Errorf("unauthorized response topic")
	}
	if command.DeviceID != deviceID {
		return 0, fmt.Errorf("command device does not match publish destination")
	}
	commandTopic, err := contracts.DeviceTopic(deviceID, contracts.KindCommands)
	if err != nil { return 0, fmt.Errorf("build command topic: %w", err) }
	payload, err := json.Marshal(command)
	if err != nil {
		return 0, fmt.Errorf("encode command: %w", err)
	}
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	// Correlation Data is copied so the packet owns stable bytes for this publish.
	ack, err := p.Manager.Publish(operation, &paho.Publish{
		Topic:   commandTopic.String(),
		QoS:     1,
		Payload: payload,
		Properties: &paho.PublishProperties{
			ContentType:     "application/json",
			ResponseTopic:   responseTopic.String(),
			CorrelationData: append([]byte(nil), correlation...),
		},
	})
	if err != nil {
		return 0, fmt.Errorf("publish command: %w", err)
	}
	if ack == nil {
		return 0, fmt.Errorf("missing QoS 1 PUBACK")
	}
	if ack.ReasonCode >= 0x80 {
		return ack.ReasonCode, fmt.Errorf("broker rejected publish: reason 0x%02x", ack.ReasonCode)
	}
	return ack.ReasonCode, nil
}

📄 internal/commands/receive.go

package commands

import (
	"fmt"
	"time"

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

func HandleMessage(service *Service, message mqtt.Message) error {
	// Validate the route and result body before matching the pending request.
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return err
	}
	if topic.Kind != contracts.KindResponses {
		return fmt.Errorf("expected a response topic")
	}
	result, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, time.Now().UTC())
	if err != nil {
		return err
	}
	_, err = service.HandleResult(topic.DeviceID, append([]byte(nil), message.CorrelationData...), result)
	return err
}

📄 internal/simulator/dedup.go

package simulator

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

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

type cachedResult struct {
	fingerprint [32]byte
	result      contracts.CommandResult
}

type Deduplicator struct {
	mu       sync.Mutex
	capacity int
	results  map[string]cachedResult
}

// NewDeduplicator bounds how many command results this simulator retains.
func NewDeduplicator(capacity int) *Deduplicator {
	if capacity < 1 {
		capacity = 1
	}
	return &Deduplicator{capacity: capacity, results: make(map[string]cachedResult)}
}

func AuthorizedResponseTopic(deviceID, requested string) bool {
	expected, err := contracts.DeviceTopic(deviceID, contracts.KindResponses)
	return err == nil && requested == expected.String()
}

// Apply executes effect at most once for a device/command ID while this process lives.
func (d *Deduplicator) Apply(command contracts.Command, effect func() (contracts.CommandResult, error)) (contracts.CommandResult, bool, error) {
	if effect == nil {
		return contracts.CommandResult{}, false, fmt.Errorf("effect is required")
	}
	encoded, err := json.Marshal(command)
	if err != nil {
		return contracts.CommandResult{}, false, fmt.Errorf("fingerprint command: %w", err)
	}
	fingerprint := sha256.Sum256(encoded)
	key := command.DeviceID + "/" + command.CommandID
	d.mu.Lock()
	defer d.mu.Unlock()
	if previous, ok := d.results[key]; ok {
		if previous.fingerprint != fingerprint {
			return contracts.CommandResult{}, false, fmt.Errorf("command ID was reused with different content")
		}
		return previous.result, true, nil
	}
	if len(d.results) >= d.capacity {
		return contracts.CommandResult{}, false, fmt.Errorf("deduplication cache is full")
	}
	result, err := effect()
	if err != nil {
		return contracts.CommandResult{}, false, err
	}
	d.results[key] = cachedResult{fingerprint: fingerprint, result: result}
	return result, false, nil
}

📄 internal/simulator/respond.go

package simulator

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"example.com/mqtt5-learning/internal/contracts"
	"example.com/mqtt5-learning/internal/mqtt"
	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

func Respond(ctx context.Context, manager *autopaho.ConnectionManager, cache *Deduplicator, deviceID string, message mqtt.Message) error {
	// Reject retained or misrouted requests before validating or applying an effect.
	if message.Retain {
		return fmt.Errorf("retained commands are rejected")
	}
	topic, err := contracts.ParseTopic(message.Topic)
	if err != nil {
		return err
	}
	if topic.Kind != contracts.KindCommands || topic.DeviceID != deviceID {
		return fmt.Errorf("command topic is not authorized for this device")
	}
	command, err := contracts.Decode[contracts.Command](message.Payload, topic, time.Now().UTC())
	if err != nil {
		return err
	}
	if !AuthorizedResponseTopic(deviceID, message.ResponseTopic) {
		return fmt.Errorf("unauthorized response topic")
	}
	// Cache the simulated result by stable command ID to suppress repeat effects.
	result, _, err := cache.Apply(command, func() (contracts.CommandResult, error) {
		return contracts.CommandResult{SchemaVersion: contracts.SchemaVersion, MessageID: "result-" + command.CommandID, DeviceID: deviceID, OccurredAt: time.Now().UTC().Format(time.RFC3339), CommandID: command.CommandID, Status: "completed"}, nil
	})
	if err != nil {
		return err
	}
	payload, err := json.Marshal(result)
	if err != nil {
		return fmt.Errorf("encode result: %w", err)
	}
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	// Reply only to the authorized destination and echo correlation bytes unchanged.
	ack, err := manager.Publish(operation, &paho.Publish{Topic: message.ResponseTopic, QoS: 1, Payload: payload, Properties: &paho.PublishProperties{ContentType: "application/json", CorrelationData: append([]byte(nil), message.CorrelationData...)}})
	if err != nil {
		return fmt.Errorf("publish response: %w", err)
	}
	if ack == nil {
		return fmt.Errorf("missing response PUBACK")
	}
	if ack.ReasonCode >= 0x80 {
		return fmt.Errorf("response publish rejected: reason 0x%02x", ack.ReasonCode)
	}
	return nil
}

📄 cmd/simulator/main.go

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"
	"example.com/mqtt5-learning/internal/simulator"
)

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()
	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 }
	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)
	// Keep recent command results so repeated delivery does not repeat the effect.
	cache := simulator.NewDeduplicator(1024)
	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	sequence := 0
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-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 := <-inbox:
			if err := simulator.Respond(ctx, manager, cache, cfg.MQTT.DeviceID, message); err != nil {
				log.Printf("command rejected or response failed topic=%q reason=%v", message.Topic, err)
			}
		}
	}
}

📄 internal/mqtt/client.go — complete file with Section 7 message properties

This target version extends the existing MQTT transport message and copies MQTT 5 response properties inside the callback, before the received packet goes out of scope.

package mqtt

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

type Message struct {
	Topic           string
	Payload         []byte
	Retain          bool
	ResponseTopic   string
	CorrelationData []byte
}

func Start(ctx context.Context, cfg config.Config, topics []contracts.TopicFilter, inbox chan<- Message) (*autopaho.ConnectionManager, error) {
	broker, err := url.Parse(cfg.MQTT.URL)
	if err != nil {
		return nil, err
	}
	initialSubscription := make(chan error, 1)
	manager, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{
		ServerUrls:                    []*url.URL{broker},
		KeepAlive:                     20,
		CleanStartOnInitialConnection: true,
		SessionExpiryInterval:         0,
		ConnectUsername:               cfg.MQTT.Username,
		ConnectPassword:               []byte(cfg.MQTT.Password),
		OnConnectionUp: func(cm *autopaho.ConnectionManager, _ *paho.Connack) {
			// This runs again on reconnect; reestablish broker subscriptions.
			subCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
			defer cancel()

			subscriptions := make([]paho.SubscribeOptions, 0, len(topics))
			for _, topic := range topics {
				subscriptions = append(subscriptions, paho.SubscribeOptions{Topic: topic.String(), QoS: 1, RetainAsPublished: true})
			}

			ack, subErr := cm.Subscribe(subCtx, &paho.Subscribe{Subscriptions: subscriptions})
			if subErr == nil {
				if ack == nil || len(ack.Reasons) != len(topics) {
					subErr = fmt.Errorf("missing or incomplete SUBACK")
				} else {
					for i, reason := range ack.Reasons {
						if reason > 1 { // We requested QoS 1; valid grants are 0 or 1.
							subErr = fmt.Errorf("subscription %s rejected: 0x%02x", topics[i], reason)
							break
						}
					}
				}
			}
			if subErr != nil {
				log.Printf("subscribe failed: %v", subErr)
			} else {
				log.Printf("subscriptions ready: %v", topics)
			}
			select {
			case initialSubscription <- subErr:
			default:
			}
		},
		OnConnectionDown:   func() bool { log.Print("disconnected; reconnecting"); return true },
		OnConnectError:     func(err error) { log.Printf("connect error: %v", err) },
		ClientID:           cfg.MQTT.ClientID,
		OnClientError:      func(err error) { log.Printf("client error: %v", err) },
		OnServerDisconnect: func(d *paho.Disconnect) { log.Printf("disconnect reason=0x%02x", d.ReasonCode) },
		OnPublishReceived: []func(paho.PublishReceived) (bool, error){
			func(received paho.PublishReceived) (bool, error) {
				packet := received.Packet
				message := Message{Topic: packet.Topic, Payload: append([]byte(nil), packet.Payload...), Retain: packet.Retain}
				if properties := packet.Properties; properties != nil {
					message.ResponseTopic = properties.ResponseTopic
					message.CorrelationData = append([]byte(nil), properties.CorrelationData...)
				}
				select {
				case inbox <- message:
				default:
					// Deliberately lossy early demo: automatic MQTT ACK is still possible.
					log.Printf("lab inbox full; message dropped topic=%s", packet.Topic)
				}
				return true, nil
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("start AutoPaho: %w", err)
	}
	startup, cancel := context.WithTimeout(ctx, cfg.ConnectTimeout())
	defer cancel()
	if err = manager.AwaitConnection(startup); err != nil {
		Stop(manager)
		return nil, fmt.Errorf("await connection: %w", err)
	}
	select {
	case err = <-initialSubscription:
		if err != nil {
			Stop(manager)
			return nil, err
		}
	case <-startup.Done():
		Stop(manager)
		return nil, fmt.Errorf("await subscriptions: %w", startup.Err())
	}
	return manager, nil
}

func Publish(ctx context.Context, manager *autopaho.ConnectionManager, topic contracts.TopicName, payload []byte) error {
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()

	ack, err := manager.Publish(operation, &paho.Publish{
		Topic:   topic.String(),
		Payload: payload,
		QoS:     1,
		Retain:  false,
		Properties: &paho.PublishProperties{
			ContentType: "application/json",
		},
	})
	if err != nil {
		return fmt.Errorf("publish: %w", err)
	}
	if ack == nil {
		return fmt.Errorf("missing Qos 1 publication response")
	}
	if ack.ReasonCode >= 0x80 {
		return fmt.Errorf("publish rejected: 0x%02x", ack.ReasonCode)
	}

	log.Printf("broker publish acknowledgement topic=%s reason=0x%02x", topic, ack.ReasonCode)

	return nil
}

func Stop(manager *autopaho.ConnectionManager) {
	shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := manager.Disconnect(shutdown); err != nil {
		log.Printf("shutdown did not finish within deadline: %v", err)
	}
}