Haystack/backend/events.go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"screenmark/screenmark/middleware"
	"strconv"

	"github.com/google/uuid"
)

const (
	IMAGE_TYPE = "image"
	LIST_TYPE  = "list"
)

type imageNotification struct {
	Type string

	ImageID   uuid.UUID
	ImageName string

	Status string
}

type listNotification struct {
	Type string

	ListID uuid.UUID
	Name   string

	Status string
}

type Notification struct {
	image *imageNotification
	list  *listNotification
}

func getImageNotification(image imageNotification) Notification {
	return Notification{
		image: &image,
	}
}

func getListNotification(list listNotification) Notification {
	return Notification{
		list: &list,
	}
}

func (n Notification) MarshalJSON() ([]byte, error) {
	if n.image != nil {
		return json.Marshal(n.image)
	}

	if n.list != nil {
		return json.Marshal(n.list)
	}

	return nil, fmt.Errorf("no image or list present")
}

func (n *Notification) UnmarshalJSON(data []byte) error {
	return fmt.Errorf("unimplemented")
}

/*
 * TODO: We have channels open every a user sends an image.
 * We never close these channels.
 *
 * What is a reasonable default? Close the channel after 1 minute of inactivity?
 */
func CreateEventsHandler(notifier *Notifier[Notification]) http.HandlerFunc {
	counter := 0

	userSplitters := make(map[string]*ChannelSplitter[Notification])

	return func(w http.ResponseWriter, r *http.Request) {
		_userId := r.Context().Value(middleware.USER_ID).(uuid.UUID)
		if _userId == uuid.Nil {
			w.WriteHeader(http.StatusUnauthorized)
			return
		}

		userId := _userId.String()

		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("Connection", "keep-alive")
		// w.(http.Flusher).Flush()

		if _, exists := notifier.Listeners[userId]; !exists {
			notifier.Create(userId)
		}

		userNotifications := notifier.Listeners[userId]

		if _, exists := userSplitters[userId]; !exists {
			splitter := NewChannelSplitter(userNotifications)

			userSplitters[userId] = &splitter
			splitter.Listen()
		}

		splitter := userSplitters[userId]

		id := strconv.Itoa(counter)
		counter += 1

		notifications := splitter.Add(id)
		defer splitter.Remove(id)

		for {
			select {
			case <-r.Context().Done():
				fmt.Fprint(w, "event: close\ndata: Connection closed\n\n")
				w.(http.Flusher).Flush()
				return
			case msg := <-notifications:

				msgString, err := json.Marshal(msg)
				if err != nil {
					w.WriteHeader(http.StatusInternalServerError)
					return
				}

				fmt.Printf("Sending msg %s\n", msgString)

				fmt.Fprintf(w, "event: data\ndata: %s\n\n", string(msgString))
				w.(http.Flusher).Flush()
			}
		}
	}
}