new-server/app/services/events.go

package services

import (
	"context"
	"fmt"
	"git.aiterp.net/lucifer/new-server/app/config"
	"git.aiterp.net/lucifer/new-server/models"
	"log"
	"math"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

func StartEventHandler() {
	config.EventChannel <- models.Event{Name: "LuciferStarted"}

	go func() {
		for event := range config.EventChannel {
			responses := handleEvent(event)

			for _, response := range responses {
				handleEvent(response)
			}
		}
	}()

	// Generate TimeChanged event
	go func() {
		drift := time.Now().Add(time.Minute).Truncate(time.Minute).Sub(time.Now())
		time.Sleep(drift + time.Millisecond*5)

		for range time.NewTicker(time.Minute).C {
			config.EventChannel <- models.Event{Name: "TimeChanged", Payload: map[string]string{
				"weekdayName": time.Now().In(loc).Format("Monday"),
				"month":       time.Now().In(loc).Format("01"),
				"year":        time.Now().In(loc).Format("2006"),
			}}
		}
	}()
}

var loc, _ = time.LoadLocation("Europe/Oslo")
var ctx = context.Background()

func handleEvent(event models.Event) (responses []models.Event) {
	var deadHandlers []models.EventHandler

	startTime := time.Now()
	defer func() {
		duration := time.Since(startTime)
		if duration > time.Millisecond*250 {
			log.Println(event.Name, "took long to handle:", duration)
		}
	}()

	if !event.HasPayload("hour") {
		event.AddPayload("hour", time.Now().In(loc).Format("15"))
	}
	if !event.HasPayload("minute") {
		event.AddPayload("minute", time.Now().In(loc).Format("04"))
	}

	err := handleSpecial(event)
	if err != nil {
		log.Printf("Special event handler error (%s): %v", event.Name, err)
		return
	}

	paramStrings := make([]string, 0, 8)
	for key, value := range event.Payload {
		paramStrings = append(paramStrings, fmt.Sprintf("%s=%s", key, value))
	}
	sort.Strings(paramStrings)

	log.Printf("Event %s(%s)", event.Name, strings.Join(paramStrings, ", "))

	handlers, err := config.EventHandlerRepository().FetchAll(ctx)
	if err != nil {
		log.Printf("Error fetchin' event halders: %s", err)
		return
	}

	allDevices := make([]models.Device, 0, 16)
	actions := make(map[int]models.EventAction, 16)
	priorities := make(map[int]int, 16)
	highestPriority := 0
	var prioritizedEvent *models.Event

	for _, handler := range handlers {
		if handler.EventName != event.Name {
			continue
		}

		devices, err := config.DeviceRepository().FetchByReference(ctx, handler.TargetKind, handler.TargetValue)
		if err != nil {
			log.Printf("Error fetchin' event devices: %s", err)
			return
		}

		if !handler.MatchesEvent(event, devices) {
			continue
		}

		if handler.OneShot {
			deadHandlers = append(deadHandlers, handler)
		}

		if handler.Priority > highestPriority {
			highestPriority = handler.Priority
			prioritizedEvent = handler.Actions.FireEvent
		}

		for _, device := range devices {
			_, ok := actions[device.ID]
			if !ok {
				allDevices = append(allDevices, device)
				actions[device.ID] = handler.Actions
				priorities[device.ID] = handler.Priority
			} else if handler.Priority > priorities[device.ID] {
				actions[device.ID] = handler.Actions
				priorities[device.ID] = handler.Priority
			}
		}
	}

	if prioritizedEvent != nil {
		responses = append(responses, *prioritizedEvent)
	}

	for i, device := range allDevices {
		action := actions[device.ID]
		newState := models.NewDeviceState{}

		newState.Color = action.SetColor
		newState.Power = action.SetPower

		if action.SetIntensity != nil {
			newState.Intensity = action.SetIntensity
		} else if action.AddIntensity != nil {
			newIntensity := math.Max(0, math.Min(1, device.State.Intensity+*action.AddIntensity))
			newState.Intensity = &newIntensity
		}

		if action.SetTemperature != nil {
			newState.Temperature = action.SetTemperature
		}

		err = allDevices[i].SetState(newState)
		if err != nil {
			log.Println("Error updating state for device", device.ID, "err:", err)
		}
	}

	config.PublishChannel <- allDevices

	wg := sync.WaitGroup{}
	for _, device := range allDevices {
		wg.Add(1)

		go func(device models.Device) {
			err := config.DeviceRepository().Save(context.Background(), &device, models.SMState)
			if err != nil {
				log.Println("Failed to save device for state:", err)
			}

			wg.Done()
		}(device)
	}
	for _, handler := range deadHandlers {
		wg.Add(1)

		go func(handler models.EventHandler) {
			err := config.EventHandlerRepository().Delete(context.Background(), &handler)
			if err != nil {
				log.Println("Failed to delete spent one-shot event handler:", err)
			}

			wg.Done()
		}(handler)
	}

	wg.Wait()

	return
}

func handleSpecial(event models.Event) error {
	switch event.Name {
	case models.ENBridgeConnected:
		bridgeId, _ := strconv.Atoi(event.Payload["bridgeId"])
		bridge, err := config.BridgeRepository().Find(ctx, bridgeId)
		if err != nil {
			return err
		}

		devices, err := config.DeviceRepository().FetchByReference(ctx, models.RKBridgeID, event.Payload["bridgeId"])
		if err != nil {
			return err
		}

		driver, err := config.DriverProvider().Provide(bridge.Driver)
		if err != nil {
			return err
		}

		return driver.Publish(ctx, bridge, devices)
	default:
		return nil
	}
}