Give stale work an expiry; treat presence as evidence

package contracts

import (
	"fmt"
	"strings"
)

type TopicName string
type TopicFilter string
type Kind string

const (
	KindTelemetry Kind = "telemetry"
	KindCommands  Kind = "commands"
	KindResponses Kind = "responses"
	KindSales     Kind = "sales"
	KindPresence  Kind = "presence"
	KindDesired   Kind = "desired"
	KindReported  Kind = "reported"
)

const topicPrefix = "lab/v1/devices/"

type Topic struct {
	DeviceID string
	Kind     Kind
}

func (t TopicName) String() string   { return string(t) }
func (f TopicFilter) String() string { return string(f) }

func DeviceTopic(deviceID string, kind Kind) (TopicName, error) {
	if !validID(deviceID) {
		return "", fmt.Errorf("invalid device ID %q", deviceID)
	}
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicName(topicPrefix + deviceID + "/" + string(kind)), nil
}

func DeviceFilter(kind Kind) (TopicFilter, error) {
	if !supportedKind(kind) {
		return "", fmt.Errorf("unsupported topic kind %q", kind)
	}
	return TopicFilter(topicPrefix + "+/" + string(kind)), nil
}

func DeviceSubscription(deviceID string, kind Kind) (TopicFilter, error) {
	topic, err := DeviceTopic(deviceID, kind)
	if err != nil {
		return "", err
	}
	return TopicFilter(topic.String()), nil
}

func ParseTopic(value string) (Topic, error) {
	if !strings.HasPrefix(value, topicPrefix) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	parts := strings.Split(strings.TrimPrefix(value, topicPrefix), "/")
	if len(parts) != 2 || !validID(parts[0]) {
		return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix)
	}
	kind := Kind(parts[1])
	if !supportedKind(kind) {
		return Topic{}, fmt.Errorf("unsupported topic kind %q", kind)
	}
	return Topic{DeviceID: parts[0], Kind: kind}, nil
}

func supportedKind(kind Kind) bool {
	switch kind {
	case KindTelemetry, KindCommands, KindResponses, KindSales,
		KindPresence, KindDesired, KindReported:
		return true
	default:
		return false
	}
}

func validID(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
}

package mqtt

import (
	"fmt"

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

// ExpiringPublish builds a publication whose broker-queued lifetime is bounded.
// The device must still check its application deadline before executing it.
func ExpiringPublish(topic contracts.TopicName, payload []byte, expirySeconds uint32) (*paho.Publish, error) {
	if topic == "" {
		return nil, fmt.Errorf("topic is required")
	}
	if expirySeconds == 0 {
		return nil, fmt.Errorf("expirySeconds must be greater than zero")
	}
	seconds := expirySeconds
	return &paho.Publish{
		Topic: topic.String(), Payload: append([]byte(nil), payload...), QoS: 1,
		Properties: &paho.PublishProperties{
			ContentType: "application/json",
			MessageExpiry: &seconds,
		},
	}, nil
}

package deviceview

import (
	"fmt"
	"time"
)

// CheckCommandDeadline rejects a command before the simulator starts its effect.
// The absolute deadline protects against time spent in local queues after delivery.
func CheckCommandDeadline(deadline string, now time.Time) error {
	if deadline == "" {
		return fmt.Errorf("command deadline is required")
	}
	when, err := time.Parse(time.RFC3339, deadline)
	if err != nil {
		return fmt.Errorf("parse command deadline: %w", err)
	}
	if !now.Before(when) {
		return fmt.Errorf("command deadline has passed")
	}
	return nil
}

package deviceview

import "sync"

type RevisionOutcome uint8

const (
	RevisionApplied RevisionOutcome = iota + 1
	RevisionDuplicate
	RevisionStale
)

type RevisionStore struct {
	mu      sync.Mutex
	current map[string]uint64
}

func NewRevisionStore() *RevisionStore {
	return &RevisionStore{current: make(map[string]uint64)}
}

func (s *RevisionStore) Apply(deviceID string, revision uint64) RevisionOutcome {
	s.mu.Lock()
	defer s.mu.Unlock()
	current, exists := s.current[deviceID]
	if exists && revision == current {
		return RevisionDuplicate
	}
	if exists && revision < current {
		return RevisionStale
	}
	s.current[deviceID] = revision
	return RevisionApplied
}

func (s *RevisionStore) Current(deviceID string) (uint64, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	revision, ok := s.current[deviceID]
	return revision, ok
}

package mqtt

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

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

type Presence struct {
	SchemaVersion int    `json:"schema_version"`
	MessageID     string `json:"message_id"`
	DeviceID      string `json:"device_id"`
	BootID        string `json:"boot_id"`
	Status        string `json:"status"`
	OccurredAt    string `json:"occurred_at"`
}

func OfflineWill(deviceID, bootID string) (*paho.WillMessage, *paho.WillProperties, error) {
	if deviceID == "" || bootID == "" {
		return nil, nil, fmt.Errorf("device ID and boot ID are required")
	}
	topic, err := contracts.DeviceTopic(deviceID, contracts.KindPresence)
	if err != nil {
		return nil, nil, fmt.Errorf("build presence topic: %w", err)
	}
	body, err := json.Marshal(Presence{
		SchemaVersion: contracts.SchemaVersionOne,
		MessageID: "will-" + bootID, DeviceID: deviceID, BootID: bootID,
		Status: "offline", OccurredAt: time.Now().UTC().Format(time.RFC3339),
	})
	if err != nil { return nil, nil, fmt.Errorf("encode offline Will: %w", err) }
	delay := uint32(5)
	return &paho.WillMessage{Topic: topic.String(), QoS: 1, Retain: true, Payload: body},
		&paho.WillProperties{WillDelayInterval: &delay, ContentType: "application/json"}, nil
}

func OnlinePresence(deviceID, bootID string, now time.Time) (*paho.Publish, error) {
	if deviceID == "" || bootID == "" {
		return nil, fmt.Errorf("device ID and boot ID are required")
	}
	topic, err := contracts.DeviceTopic(deviceID, contracts.KindPresence)
	if err != nil { return nil, fmt.Errorf("build presence topic: %w", err) }
	body, err := json.Marshal(Presence{
		SchemaVersion: contracts.SchemaVersionOne,
		MessageID: "online-" + bootID, DeviceID: deviceID, BootID: bootID,
		Status: "online", OccurredAt: now.UTC().Format(time.RFC3339),
	})
	if err != nil { return nil, fmt.Errorf("encode online presence: %w", err) }
	return &paho.Publish{
		Topic: topic.String(), Payload: body, QoS: 1, Retain: true,
		Properties: &paho.PublishProperties{ContentType: "application/json"},
	}, nil
}

1. Broker expiry limits onward delivery; an application deadline limits whether the device starts the action.
2. Retained messages represent latest state, not historical event delivery.
3. Heartbeat freshness says a recent report arrived, not that the device is healthy or the broker currently has a connection.