Ryan
fb30e8e884
Move lesson source into the teaching structure, align the course path with the learner's Go and DevOps experience, and add the interactive lesson guides. Add advanced DevOps lifecycle study as a separate section. Bound MQTT message work and drain accepted jobs during shutdown; align the backend implementation with the revised course through Section 6.
97 lines
2.7 KiB
Go
97 lines
2.7 KiB
Go
package deviceview
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"example.com/mqtt5-learning/internal/contracts"
|
|
"example.com/mqtt5-learning/internal/mqtt"
|
|
)
|
|
|
|
type Service struct {
|
|
store *Store
|
|
logger *slog.Logger
|
|
freshFor time.Duration
|
|
now func() time.Time
|
|
}
|
|
|
|
func NewService(store *Store, logger *slog.Logger, freshFor time.Duration) (*Service, error) {
|
|
if store == nil {
|
|
return nil, fmt.Errorf("device-view store is required")
|
|
}
|
|
if logger == nil {
|
|
return nil, fmt.Errorf("logger is required")
|
|
}
|
|
if freshFor <= 0 {
|
|
return nil, fmt.Errorf("heartbeat freshness window must be positive")
|
|
}
|
|
return &Service{store: store, logger: logger, freshFor: freshFor, now: time.Now}, nil
|
|
}
|
|
|
|
func (s *Service) HandleMessage(ctx context.Context, message mqtt.Message) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
topic, err := contracts.ParseTopic(message.Topic)
|
|
if err != nil {
|
|
return fmt.Errorf("parse topic: %w", err)
|
|
}
|
|
receivedAt := s.now().UTC()
|
|
switch topic.Kind {
|
|
case contracts.KindTelemetry:
|
|
heartbeat, err := contracts.Decode[contracts.Heartbeat](message.Payload, topic, receivedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("validate heartbeat: %w", err)
|
|
}
|
|
updated, err := s.store.RecordHeartbeat(heartbeat, receivedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("record heartbeat: %w", err)
|
|
}
|
|
if !updated {
|
|
s.logger.Info("older or duplicate heartbeat ignored",
|
|
"device_id", heartbeat.DeviceID,
|
|
"message_id", heartbeat.MessageID,
|
|
)
|
|
return nil
|
|
}
|
|
s.logger.Info("heartbeat recorded",
|
|
"device_id", heartbeat.DeviceID,
|
|
"message_id", heartbeat.MessageID,
|
|
"received_at", receivedAt,
|
|
)
|
|
return nil
|
|
case contracts.KindSales:
|
|
report, err := contracts.Decode[contracts.SalesReport](message.Payload, topic, receivedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("validate sales report: %w", err)
|
|
}
|
|
// Validation is not durable admission; Section 13 stores reports.
|
|
s.logger.Info("sales report validated; persistence not implemented",
|
|
"device_id", report.DeviceID,
|
|
"report_id", report.ReportID,
|
|
"item_count", len(report.Items),
|
|
)
|
|
return nil
|
|
case contracts.KindResponses:
|
|
result, err := contracts.Decode[contracts.CommandResult](message.Payload, topic, receivedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("validate command result: %w", err)
|
|
}
|
|
// Section 7 adds pending-command and correlation matching.
|
|
s.logger.Info("command result validated; matching not implemented",
|
|
"device_id", result.DeviceID,
|
|
"command_id", result.CommandID,
|
|
"status", result.Status,
|
|
)
|
|
return nil
|
|
default:
|
|
return fmt.Errorf("device-view service does not handle topic kind %q", topic.Kind)
|
|
}
|
|
}
|
|
|
|
func (s *Service) Status(deviceID string) (Status, bool) {
|
|
return s.store.GetStatus(deviceID, s.now().UTC(), s.freshFor)
|
|
}
|