home_automatization/services/device-control-service/internal/shadow/store.go

// 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
}