display.go

package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"sync"

	"github.com/JamesStewy/sse"
	"github.com/hashicorp/go-multierror"

	"stream-server/internal/util"
)

// displayState struct defines the state of the display.
type displayState struct {
	LargeChannel string  `json:"large_channel"`
	SmallChannel string  `json:"small_channel"`
	Volume       float32 `json:"volume"`
	SmallScale   float32 `json:"small_scale"`
	ShowChat     bool    `json:"show_chat"`
}

// Display struct defines the Display itself
type Display struct {
	State     displayState
	clientsMu sync.Mutex
	// clients holds the clients that are connected to the event handler. It is used to broadcast state changes to all SSE (Server-Sent Events) clients.
	// Must acquire clientsMu before accessing the map.
	// Note: the only reason we use a sse.Client => bool map is that we can call *delete* with the client as key. The actual bool value that is stored holds no significance whatsoever.
	// This is basically a *set* in go.
	clients map[*sse.Client]bool
	// the prefix used for logging
	loggingPrefix string
}

// NewDisplay creates a new Display with default values
func NewDisplay(namespace string) *Display {
	var loggingPrefix string
	if namespace == "default" {
		loggingPrefix = "display"
	} else {
		loggingPrefix = fmt.Sprintf("display[%s]", namespace)
	}

	return &Display{
		State: displayState{
			LargeChannel: "",
			SmallChannel: "",
			Volume:       0.5,
			SmallScale:   0.3,
			ShowChat:     false,
		},
		clients:       make(map[*sse.Client]bool),
		loggingPrefix: loggingPrefix,
	}
}

// stateHandler updated the state based on the query string.
// For example
// 	/display?large_channel=asdf&small_channel=null&volume=&small_scale=0.25&show_chat=true
// will
//  - set large_channel to asdf
//  - reset small_channel to null
//  - do nothing with volume (the parameter can also be omitted)
//  - set small_scale to 0.25
//  - set show_chat to true
//
// If the floats (volume and small_scale) or the boolean (show_chat) could not be parsed, no changes are made at all and an error gets returned even if other fields could be parsed.
func (d *Display) stateHandler(w http.ResponseWriter, r *http.Request) {
	// the new state we will replace the current state with IF everything parses correctly.
	newState := d.State

	// we save the error messages of the 5 parsing steps here.
	var errs error

	if value, err := util.GetSingleQueryParam(r, "large_channel"); value != nil {
		if err != nil {
			errs = multierror.Append(err)
		} else {
			newState.LargeChannel = *value
		}
	}

	if value, err := util.GetSingleQueryParam(r, "small_channel"); value != nil {
		if err != nil {
			errs = multierror.Append(err)
		} else {
			newState.SmallChannel = *value
		}
	}

	if value, err := util.GetSingleQueryParam(r, "volume"); value != nil {
		if err != nil {
			errs = multierror.Append(err)
		} else {
			newVolume, err := strconv.ParseFloat(*value, 32)
			if err != nil {
				errs = multierror.Append(errs, err)
			} else {
				newState.Volume = float32(newVolume)
			}
		}
	}

	if value, err := util.GetSingleQueryParam(r, "small_scale"); value != nil {
		if err != nil {
			errs = multierror.Append(err)
		} else {
			newSmallScale, err := strconv.ParseFloat(*value, 32)
			if err != nil {
				errs = multierror.Append(errs, err)
			} else {
				newState.SmallScale = float32(newSmallScale)
			}
		}
	}

	if value, err := util.GetSingleQueryParam(r, "show_chat"); value != nil {
		if err != nil {
			errs = multierror.Append(err)
		} else {
			newShowChat, err := strconv.ParseBool(*value)
			if err != nil {
				errs = multierror.Append(errs, err)
			} else {
				newState.ShowChat = newShowChat
			}
		}
	}

	if errs != nil { // we had errors
		http.Error(w, errs.Error(), http.StatusBadRequest)

		logger.Printf("%s: Error(s) while parsing /display query: %v\n", d.loggingPrefix, errs)
		return
	}

	if d.State != newState {
		d.State = newState

		body, err := json.Marshal(d.State)
		if err != nil {
			return
		}
		logger.Printf("%s: Updated (%s):\n%s\n", d.loggingPrefix, r.URL.RawQuery, string(body))

		// craft the SSE message
		msg := sse.Msg{
			Data: string(body),
		}

		// We'd like to broadcast the message to all clients, as
		// recorded in the clients map. The message send is a
		// potentially blocking operation so we don't want to hold the
		// lock across send operations. Thus, we first build up a list
		// of clients to send to (while holding clientsMu) and start
		// sending only after giving up the lock again.
		var sendingTo []*sse.Client
		d.clientsMu.Lock()
		for client := range d.clients {
			sendingTo = append(sendingTo, client)
		}
		d.clientsMu.Unlock()

		// now broadcast it to all clients
		for _, client := range sendingTo {
			client.Send(msg)
		}
	}

	util.ServeIndentedJSON(w, r, d.State)
}

// eventsHandler serves a SSE (Server-sent events) endpoint that the website(s) can connect to. It publishes the state when it gets changed.
func (d *Display) eventsHandler(w http.ResponseWriter, r *http.Request) {
	client, err := sse.ClientInit(w)

	// return error if unable to initialize sse connection
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// Add client to broadcast set. Need to serialize access to clients as
	// we're running concurrently with other requests.
	d.clientsMu.Lock()
	d.clients[client] = true
	d.clientsMu.Unlock()

	// remove client from broadcast set on exit
	defer func() {
		d.clientsMu.Lock()
		delete(d.clients, client)
		d.clientsMu.Unlock()
	}()

	body, err := json.Marshal(d.State)
	if err != nil {
		return
	}

	go func() {
		// client.Send will block the channel that client.Run is listening on
		// thats why we send the initial state in an goroutine that runs concurrently with client.Run
		client.Send(sse.Msg{
			Data: string(body),
		})
	}()

	// run the in the context of the request
	client.Run(r.Context())
}