// Package shadow implements the Device Shadow pattern on top of Redis: // desired_state (what the user wants) vs reported_state (what the device // confirmed), plus status and last_seen for liveness tracking. package shadow import ( "context" "encoding/json" "fmt" "strconv" "time" "github.com/redis/go-redis/v9" ) // knownDevicesKey backs a lightweight device registry: health-check needs to // know which device IDs to scan, and this service has no PostgreSQL access // (device registration lives in Laravel), so every device we hear from gets // added to this set. const knownDevicesKey = "devices:known" func desiredKey(deviceID string) string { return "device:" + deviceID + ":desired_state" } func reportedKey(deviceID string) string { return "device:" + deviceID + ":reported_state" } func statusKey(deviceID string) string { return "device:" + deviceID + ":status" } func lastSeenKey(deviceID string) string { return "device:" + deviceID + ":last_seen" } const ( StatusOnline = "online" StatusOffline = "offline" ) type Store struct { rdb *redis.Client } func New(rdb *redis.Client) *Store { return &Store{rdb: rdb} } func Connect(ctx context.Context, addr, password string, db int) (*Store, error) { rdb := redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db}) if err := rdb.Ping(ctx).Err(); err != nil { return nil, fmt.Errorf("ping redis: %w", err) } return New(rdb), nil } func (s *Store) Close() error { return s.rdb.Close() } // PatchDesiredState merges patch into the device's desired_state JSON object. func (s *Store) PatchDesiredState(ctx context.Context, deviceID string, patch map[string]any) error { return s.patchJSON(ctx, desiredKey(deviceID), patch) } // PatchReportedState merges patch into the device's reported_state JSON object. func (s *Store) PatchReportedState(ctx context.Context, deviceID string, patch map[string]any) error { return s.patchJSON(ctx, reportedKey(deviceID), patch) } func (s *Store) patchJSON(ctx context.Context, key string, patch map[string]any) error { existing := map[string]any{} raw, err := s.rdb.Get(ctx, key).Result() switch { case err == redis.Nil: // no existing state yet, start from an empty object case err != nil: return fmt.Errorf("get %s: %w", key, err) default: if err := json.Unmarshal([]byte(raw), &existing); err != nil { return fmt.Errorf("unmarshal existing %s: %w", key, err) } } for k, v := range patch { existing[k] = v } merged, err := json.Marshal(existing) if err != nil { return fmt.Errorf("marshal %s: %w", key, err) } if err := s.rdb.Set(ctx, key, merged, 0).Err(); err != nil { return fmt.Errorf("set %s: %w", key, err) } return nil } // Touch registers deviceID as known, bumps last_seen to now, and — if the // device wasn't already marked online — flips status to online. It reports // whether that online transition happened, so the caller can decide whether // a "device back online" event is worth emitting. func (s *Store) Touch(ctx context.Context, deviceID string) (becameOnline bool, err error) { if err := s.rdb.SAdd(ctx, knownDevicesKey, deviceID).Err(); err != nil { return false, fmt.Errorf("register known device %s: %w", deviceID, err) } if err := s.rdb.Set(ctx, lastSeenKey(deviceID), time.Now().Unix(), 0).Err(); err != nil { return false, fmt.Errorf("set last_seen for %s: %w", deviceID, err) } status, err := s.rdb.Get(ctx, statusKey(deviceID)).Result() if err != nil && err != redis.Nil { return false, fmt.Errorf("get status for %s: %w", deviceID, err) } if status == StatusOnline { return false, nil } if err := s.rdb.Set(ctx, statusKey(deviceID), StatusOnline, 0).Err(); err != nil { return false, fmt.Errorf("set status online for %s: %w", deviceID, err) } return true, nil } // MarkOfflineIfStale flips deviceID to offline if it is currently online and // its last_seen is older than timeout. It reports whether a change was made. func (s *Store) MarkOfflineIfStale(ctx context.Context, deviceID string, timeout time.Duration) (changed bool, err error) { status, err := s.rdb.Get(ctx, statusKey(deviceID)).Result() if err == redis.Nil || status != StatusOnline { return false, nil } if err != nil { return false, fmt.Errorf("get status for %s: %w", deviceID, err) } lastSeenRaw, err := s.rdb.Get(ctx, lastSeenKey(deviceID)).Result() if err != nil && err != redis.Nil { return false, fmt.Errorf("get last_seen for %s: %w", deviceID, err) } var lastSeen time.Time if lastSeenRaw != "" { sec, err := strconv.ParseInt(lastSeenRaw, 10, 64) if err != nil { return false, fmt.Errorf("parse last_seen for %s: %w", deviceID, err) } lastSeen = time.Unix(sec, 0) } if time.Since(lastSeen) <= timeout { return false, nil } if err := s.rdb.Set(ctx, statusKey(deviceID), StatusOffline, 0).Err(); err != nil { return false, fmt.Errorf("set status offline for %s: %w", deviceID, err) } return true, nil } // KnownDevices lists every device ID this instance has ever heard from. func (s *Store) KnownDevices(ctx context.Context) ([]string, error) { ids, err := s.rdb.SMembers(ctx, knownDevicesKey).Result() if err != nil { return nil, fmt.Errorf("list known devices: %w", err) } return ids, nil }