Connect MQTT patterns to vending workflows

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 &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 < 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 && catalog.Revision <= 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) > 64 || len(value) > 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 && config.Revision <= 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" && status != "warning" && 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 < 0x20 && r != '\t' { return -1 }
		return r
	}, message)
	clean = strings.TrimSpace(clean)
	if len(clean) > 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
}

1. Use stable application report identity, not MQTT packet ID.
2. Device-reported applied revision is application evidence; PUBACK is transport evidence.
3. Fresh heartbeat, current MQTT connection, and device health measure different facts.