underbbs/server/server.go

package server

import (
	"context"
	"errors"
	"fmt"
	"forge.lightcrystal.systems/lightcrystal/underbbs/adapter"
	"forge.lightcrystal.systems/lightcrystal/underbbs/models"
	"golang.org/x/time/rate"
	"hacklab.nilfm.cc/quartzgun/cookie"
	"hacklab.nilfm.cc/quartzgun/renderer"
	"io/ioutil"
	"log"
	"net/http"
	"nhooyr.io/websocket"
	"sync"
	"time"
)

type Subscriber struct {
	key       string
	msgs      chan []byte
	data      chan models.SocketData
	closeSlow func()
}

type BBSServer struct {
	subscribeMessageBuffer int
	publishLimiter         *rate.Limiter
	logf                   func(f string, v ...interface{})
	serveMux               http.ServeMux
	subscribersLock        sync.Mutex
	subscribers            map[*Subscriber][]adapter.Adapter
}

func New() *BBSServer {
	srvr := &BBSServer{
		subscribeMessageBuffer: 16,
		logf:                   log.Printf,
		subscribers:            make(map[*Subscriber][]adapter.Adapter),
	}

	// frontend is here
	srvr.serveMux.Handle("/app/", http.StripPrefix("/app/", renderer.Subtree("./dist")))

	// api
	srvr.serveMux.Handle("/api/", http.StripPrefix("/api", srvr.apiMux()))

	// websocket
	srvr.serveMux.HandleFunc("/subscribe", srvr.subscribeHandler)
	srvr.serveMux.HandleFunc("/publish", srvr.publishHandler)

	return srvr
}

func (self *BBSServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	self.serveMux.ServeHTTP(w, r)
}

func (self *BBSServer) subscribeHandler(w http.ResponseWriter, r *http.Request) {

	c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
		Subprotocols: []string{},
	})
	if err != nil {
		self.logf("%v", err)
		return
	}

	ctx := r.Context()
	ctx = c.CloseRead(ctx)

	s := &Subscriber{
		key:  cookie.GenToken(64),
		msgs: make(chan []byte, self.subscribeMessageBuffer),
		data: make(chan models.SocketData),
		closeSlow: func() {
			c.Close(websocket.StatusPolicyViolation, "connection too slow to keep up with messages")
		},
	}

	// start with an empty set of adapters
	// we'll configure them separately
	adapters := make([]adapter.Adapter, 0, 4)
	self.addSubscriber(s, adapters)

	// defer cleanup and write messages in the background
	defer self.deleteSubscriber(s)
	defer c.Close(websocket.StatusInternalError, "")

	go func() {
		fmt.Println("waiting for data on the subscriber's channel")
		for {
			select {
			case msg := <-s.msgs:
				writeTimeout(ctx, time.Second*5, c, msg)

			case <-ctx.Done():
				fmt.Println("subscriber has disconnected")
				return //ctx.Err()
			}
		}
	}()

	// give user their key
	s.msgs <- []byte("{ \"key\":\"" + s.key + "\" }")

	// block on the data channel, serializing and passing the data to the subscriber
	listen([]chan models.SocketData{s.data}, s.msgs)

	if errors.Is(err, context.Canceled) {
		return
	}
	if websocket.CloseStatus(err) == websocket.StatusNormalClosure ||
		websocket.CloseStatus(err) == websocket.StatusGoingAway {
		return
	}
	if err != nil {
		self.logf("%v", err)
		return
	}
}

func listen(channels []chan models.SocketData, out chan []byte) {
	var wg sync.WaitGroup
	for _, ch := range channels {
		wg.Add(1)
		go func(ch <-chan models.SocketData) {
			defer wg.Done()
			for data := range ch {
				out <- data.ToDatagram()
			}
		}(ch)

	}
	wg.Wait()
	close(out)
}

func (self *BBSServer) publishHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != "POST" {
		http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
		return
	}
	body := http.MaxBytesReader(w, r.Body, 8192)
	msg, err := ioutil.ReadAll(body)
	if err != nil {
		http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge)
		return
	}

	self.publish(msg)

	w.WriteHeader(http.StatusAccepted)
}

func (self *BBSServer) publish(msg []byte) {
	self.subscribersLock.Lock()
	defer self.subscribersLock.Unlock()

	// send messages to our adapter(s)

	self.publishLimiter.Wait(context.Background())

	// send any response from the adapter(s) back to the client

	/*for s, k := range self.subscribers {
		// whatever logic to select which subscriber to send back to
	}*/

}

func (self *BBSServer) addSubscriber(s *Subscriber, k []adapter.Adapter) {
	self.subscribersLock.Lock()
	self.subscribers[s] = k
	self.subscribersLock.Unlock()
}

func (self *BBSServer) deleteSubscriber(s *Subscriber) {
	self.subscribersLock.Lock()
	delete(self.subscribers, s)
	self.subscribersLock.Unlock()
}

func writeTimeout(ctx context.Context, timeout time.Duration, c *websocket.Conn, msg []byte) error {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	return c.Write(ctx, websocket.MessageText, msg)
}