Server/services/hue/bridge.go

package hue

import (
	"context"
	"errors"
	lucifer3 "git.aiterp.net/lucifer3/server"
	"git.aiterp.net/lucifer3/server/device"
	"git.aiterp.net/lucifer3/server/events"
	"git.aiterp.net/lucifer3/server/internal/gentools"
	"log"
	"strings"
	"sync"
	"time"
)

func NewBridge(host string, client *Client) *Bridge {
	return &Bridge{
		client:        client,
		host:          host,
		ctx:           context.Background(),
		cancel:        func() {},
		resources:     map[string]*ResourceData{},
		activeStates:  map[string]device.State{},
		desiredStates: map[string]device.State{},
		hasSeen:       map[string]bool{},
	}
}

type Bridge struct {
	mu sync.Mutex

	client *Client
	host   string
	ctx    context.Context
	cancel context.CancelFunc

	resources     map[string]*ResourceData
	activeStates  map[string]device.State
	desiredStates map[string]device.State
	hasSeen       map[string]bool

	lastDiscoverCancel context.CancelFunc
}

func (b *Bridge) SearchDevices(timeout time.Duration) error {
	discoverCtx, cancel := context.WithCancel(b.ctx)

	b.mu.Lock()
	if b.lastDiscoverCancel != nil {
		b.lastDiscoverCancel()
	}
	b.lastDiscoverCancel = cancel
	b.mu.Unlock()

	if timeout <= time.Second*10 {
		timeout = time.Second * 10
	}

	// Spend half the time waiting for devices
	// TODO: Wait for v2 endpoint
	ctx, cancel := context.WithTimeout(discoverCtx, timeout/2)
	defer cancel()
	err := b.client.LegacyDiscover(ctx, "sensors")
	if err != nil {
		return err
	}
	<-ctx.Done()
	if discoverCtx.Err() != nil {
		return discoverCtx.Err()
	}

	// Spend half the time waiting for lights
	// TODO: Wait for v2 endpoint
	ctx, cancel = context.WithTimeout(discoverCtx, timeout/2)
	defer cancel()
	err = b.client.LegacyDiscover(ctx, "sensors")
	if err != nil {
		return err
	}
	<-ctx.Done()
	if discoverCtx.Err() != nil {
		return discoverCtx.Err()
	}

	// Let the main loop get the new light.
	return nil
}

func (b *Bridge) RefreshAll() ([]lucifer3.Event, error) {
	ctx, cancel := context.WithTimeout(b.ctx, time.Second*15)
	defer cancel()

	allResources, err := b.client.AllResources(ctx)
	if err != nil {
		return nil, err
	}

	resources := make(map[string]*ResourceData, len(allResources))
	for i := range allResources {
		resources[allResources[i].ID] = &allResources[i]
	}

	newEvents := make([]lucifer3.Event, 0, 0)
	b.mu.Lock()
	for id, res := range resources {
		if res.Type == "device" {
			hwState, hwEvent := res.GenerateEvent(b.host, resources)
			if !b.hasSeen[id] {
				newEvents = append(newEvents, hwState, hwEvent, events.DeviceReady{ID: hwState.ID})
				b.hasSeen[id] = true
			} else {
				newEvents = append(newEvents, hwState)
			}
		}
	}
	b.resources = resources
	b.mu.Unlock()

	return newEvents, nil
}

func (b *Bridge) ApplyPatches(resources []ResourceData) (events []lucifer3.Event, shouldRefresh bool) {
	b.mu.Lock()
	mapCopy := gentools.CopyMap(b.resources)
	b.mu.Unlock()

	for _, resource := range resources {
		if mapCopy[resource.ID] != nil {
			mapCopy[resource.ID] = mapCopy[resource.ID].WithPatch(resource)
		} else {
			log.Println(resource.ID, resource.Type, "not seen!")
			shouldRefresh = true
		}
	}

	for _, resource := range resources {
		if resource.Owner != nil && resource.Owner.Kind == "device" {
			if parent, ok := mapCopy[resource.Owner.ID]; ok {
				hwState, _ := parent.GenerateEvent(b.host, mapCopy)
				events = append(events, hwState)
			}
		}
	}

	b.mu.Lock()
	b.resources = mapCopy
	b.mu.Unlock()

	return
}

func (b *Bridge) SetStates(patch map[string]device.State) {
	b.mu.Lock()
	newStates := gentools.CopyMap(b.desiredStates)
	resources := b.resources
	b.mu.Unlock()

	prefix := "hue:" + b.host + ":"
	for id, state := range patch {
		if !strings.HasPrefix(id, prefix) {
			continue
		}
		id = id[len(prefix):]

		resource := resources[id]
		if resource == nil {
			continue
		}

		newStates[id] = resource.FixState(state, resources)
	}

	b.mu.Lock()
	b.desiredStates = newStates
	b.mu.Unlock()
}

func (b *Bridge) Run(ctx context.Context, bus *lucifer3.EventBus) interface{} {
	hwEvents, err := b.RefreshAll()
	if err != nil {
		return nil
	}

	bus.RunEvents(hwEvents)

	sse := b.client.SSE(ctx)
	step := time.NewTicker(time.Second * 30)
	defer step.Stop()

	for {
		select {
		case updates, ok := <-sse:
			{
				if !ok {
					return errors.New("SSE lost connection")
				}

				newEvents, shouldUpdate := b.ApplyPatches(
					gentools.Flatten(gentools.Map(updates, func(update SSEUpdate) []ResourceData {
						return update.Data
					})),
				)

				bus.RunEvents(newEvents)

				if shouldUpdate {
					hwEvents, err := b.RefreshAll()
					if err != nil {
						return nil
					}

					bus.RunEvents(hwEvents)
				}
			}
		case <-ctx.Done():
			{
				return nil
			}
		}
	}

}