Server/services/httpapiv1/service.go

package httpapiv1

import (
	lucifer3 "git.aiterp.net/lucifer3/server"
	"git.aiterp.net/lucifer3/server/commands"
	"git.aiterp.net/lucifer3/server/effects"
	"git.aiterp.net/lucifer3/server/events"
	"git.aiterp.net/lucifer3/server/internal/color"
	"git.aiterp.net/lucifer3/server/internal/gentools"
	"git.aiterp.net/lucifer3/server/services/script"
	"git.aiterp.net/lucifer3/server/services/uistate"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"log"
	"net"
	"net/http"
	"sync"
	"time"
)

var zeroUUID = uuid.UUID{
	0, 0, 0, 0,
	0, 0, 0, 0,
	0, 0, 0, 0,
	0, 0, 0, 0,
}

func New(addr string) (lucifer3.Service, error) {
	svc := &service{}
	upgrader := websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}

	e := echo.New()
	e.HideBanner = true
	e.HidePort = true
	e.Use(middleware.CORS())

	e.GET("/color/:value", func(c echo.Context) error {
		col, err := color.Parse(c.Param("value"))
		if err != nil {
			return c.String(400, err.Error())
		}

		rgb, _ := col.ToRGB()
		xy, _ := col.ToXY()
		hs, _ := col.ToHS()

		return c.JSON(200, color.Color{
			K:   col.K,
			RGB: rgb.RGB,
			HS:  hs.HS,
			XY:  xy.XY,
		})
	})

	e.GET("/state", func(c echo.Context) error {
		svc.mu.Lock()
		data := svc.data
		svc.mu.Unlock()

		return c.JSON(200, data)
	})

	e.GET("/subscribe", func(c echo.Context) error {
		ws, err := upgrader.Upgrade(c.Response(), c.Request(), nil)
		if err != nil {
			return err
		}

		sub := make(chan uistate.Patch, 64)
		svc.addSub(sub)

		defer svc.removeSub(sub)
		defer ws.Close()

		lastSent := time.Now()
		patches := make([]uistate.Patch, 0, 128)
		for {
			patch := <-sub
			since := time.Now().Sub(lastSent)
			patches = append(patches[:0], patch)

			if since < time.Millisecond*950 {
				waitCh := time.NewTimer(time.Millisecond*1000 - since)
				waiting := true

				for waiting {
					select {
					case patch, ok := <-sub:
						patches = append(patches, patch)
						waiting = ok
					case <-waitCh.C:
						waiting = false
					}
				}
			}

			err := ws.WriteJSON(patches)
			if err != nil {
				break
			}

			lastSent = time.Now()
		}

		return nil
	})

	e.GET("/subscribe-simple", func(c echo.Context) error {
		type ChangedPatch struct {
			Devices     map[string]*uistate.Device        `json:"devices"`
			Assignments map[uuid.UUID]*uistate.Assignment `json:"assignments"`
			Scripts     map[string][]script.Line          `json:"scripts"`
			Full        *uistate.Data                     `json:"full,omitempty"`
		}

		ws, err := upgrader.Upgrade(c.Response(), c.Request(), nil)
		if err != nil {
			return err
		}

		sub := make(chan uistate.Patch, 64)
		svc.addSub(sub)

		defer svc.removeSub(sub)
		defer ws.Close()

		svc.mu.Lock()
		err = ws.WriteJSON(ChangedPatch{Full: &svc.data})
		svc.mu.Unlock()
		if err != nil {
			return err
		}

		patches := make([]uistate.Patch, 0, 128)
		for {
			patch := <-sub
			patches = append(patches[:0], patch)
			waitCh := time.After(time.Millisecond * 330)
			waiting := true
			for waiting {
				select {
				case patch, ok := <-sub:
					patches = append(patches, patch)
					waiting = ok
				case <-waitCh:
					waiting = false
				}
			}

			statePatch := ChangedPatch{
				Devices:     make(map[string]*uistate.Device),
				Assignments: make(map[uuid.UUID]*uistate.Assignment),
				Scripts:     make(map[string][]script.Line),
				Full:        nil,
			}

			svc.mu.Lock()
			for _, patch := range patches {
				if patch.Device != nil {
					if patch.Device.Delete {
						statePatch.Devices[patch.Device.ID] = nil
					} else {
						statePatch.Devices[patch.Device.ID] = gentools.Ptr(svc.data.Devices[patch.Device.ID])
					}
				}
				if patch.Assignment != nil {
					if patch.Assignment.Delete {
						statePatch.Assignments[patch.Assignment.ID] = nil
					} else {
						statePatch.Assignments[patch.Assignment.ID] = gentools.Ptr(svc.data.Assignments[patch.Assignment.ID])
					}
				}
				if patch.Script != nil {
					if len(patch.Script.Lines) > 0 {
						statePatch.Scripts[patch.Script.Name] = svc.data.Scripts[patch.Script.Name]
					} else {
						statePatch.Scripts[patch.Script.Name] = nil
					}
				}
			}
			svc.mu.Unlock()

			err := ws.WriteJSON(statePatch)
			if err != nil {
				break
			}
		}

		return nil
	})

	e.POST("/command", func(c echo.Context) error {
		var input commandInput
		err := c.Bind(&input)
		if err != nil {
			return err
		}

		svc.mu.Lock()
		bus := svc.bus
		svc.mu.Unlock()
		if bus == nil {
			return c.String(413, "Waiting for bus")
		}

		switch {
		case input.Assign != nil:
			bus.RunCommand(commands.Assign{
				ID:     nil,
				Match:  input.Assign.Match,
				Effect: input.Assign.Effect.Effect,
			})
		case input.PairDevice != nil:
			bus.RunCommand(*input.PairDevice)
		case input.ConnectDevice != nil:
			bus.RunCommand(*input.ConnectDevice)
		case input.ForgetDevice != nil:
			bus.RunCommand(*input.ForgetDevice)
		case input.SearchDevices != nil:
			bus.RunCommand(*input.SearchDevices)
		case input.AddAlias != nil:
			bus.RunCommand(*input.AddAlias)
		case input.RemoveAlias != nil:
			bus.RunCommand(*input.RemoveAlias)
		case input.UpdateScript != nil:
			bus.RunCommand(*input.UpdateScript)
		case input.ExecuteScript != nil:
			bus.RunCommand(*input.ExecuteScript)
		case input.UpdateTrigger != nil:
			if input.UpdateTrigger.ID == zeroUUID {
				input.UpdateTrigger.ID = uuid.New()
			}
			if input.UpdateTrigger.ScriptPre == nil {
				input.UpdateTrigger.ScriptPre = make([]script.Line, 0)
			}
			if input.UpdateTrigger.ScriptPost == nil {
				input.UpdateTrigger.ScriptPost = make([]script.Line, 0)
			}
			bus.RunCommand(*input.UpdateTrigger)
		case input.DeleteTrigger != nil:
			bus.RunCommand(*input.DeleteTrigger)
		default:
			return c.String(400, "No supported command found in input")
		}

		return c.JSON(200, input)
	})

	listener, err := net.Listen("tcp", addr)
	if err != nil {
		return nil, err
	}
	e.Listener = listener

	go func() {
		err := e.Start(addr)
		if err != nil {
			log.Fatalln("Failed to listen to webserver")
		}
	}()

	return svc, nil
}

type service struct {
	mu   sync.Mutex
	data uistate.Data
	bus  *lucifer3.EventBus
	subs []chan uistate.Patch
}

func (s *service) Active() bool {
	return true
}

func (s *service) HandleEvent(bus *lucifer3.EventBus, event lucifer3.Event) {
	switch event := event.(type) {
	case events.Started:
		s.mu.Lock()
		s.bus = bus
		s.mu.Unlock()
	case uistate.Patch:
		s.mu.Lock()
		s.data = s.data.WithPatch(event)
		subs := s.subs
		s.mu.Unlock()

		for _, sub := range subs {
			select {
			case sub <- event:
			default:
			}
		}
	}
}

func (s *service) addSub(ch chan uistate.Patch) {
	s.mu.Lock()
	s.subs = append(s.subs, ch)
	s.mu.Unlock()
}

func (s *service) removeSub(ch chan uistate.Patch) {
	s.mu.Lock()
	s.subs = append([]chan uistate.Patch{}, s.subs...)
	for i, sub := range s.subs {
		if sub == ch {
			s.subs = append(s.subs[:i], s.subs[i+1:]...)
			break
		}
	}
	s.mu.Unlock()
}

type commandInput struct {
	Assign        *assignInput            `json:"assign,omitempty"`
	AddAlias      *commands.AddAlias      `json:"addAlias,omitempty"`
	RemoveAlias   *commands.RemoveAlias   `json:"removeAlias,omitempty"`
	PairDevice    *commands.PairDevice    `json:"pairDevice,omitempty"`
	ConnectDevice *commands.ConnectDevice `json:"connectDevice,omitempty"`
	SearchDevices *commands.SearchDevices `json:"searchDevices,omitempty"`
	ForgetDevice  *commands.ForgetDevice  `json:"forgetDevice,omitempty"`
	UpdateScript  *script.Update          `json:"updateScript,omitempty"`
	ExecuteScript *script.Execute         `json:"executeScript,omitempty"`
	UpdateTrigger *script.UpdateTrigger   `json:"updateTrigger,omitempty"`
	DeleteTrigger *script.DeleteTrigger   `json:"deleteTrigger,omitempty"`
}

type assignInput struct {
	Match  string               `json:"match"`
	Effect effects.Serializable `json:"effect"`
}