underbbs/server/server.go

package server

import (
	"context"
	"crypto"
	"crypto/x509"
	"encoding/pem"
	"errors"
	"forge.lightcrystal.systems/nilix/quartzgun/cookie"
	"forge.lightcrystal.systems/nilix/quartzgun/renderer"
	"forge.lightcrystal.systems/nilix/underbbs/adapter"
	"forge.lightcrystal.systems/nilix/underbbs/models"
	_ "github.com/rs/cors"
	"golang.org/x/time/rate"
	"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
	apKey                  *crypto.PrivateKey
	apDomain               *string
}

func CORS(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Add("Access-Control-Allow-Origin", "*")
		w.Header().Add("Access-Control-Allow-Credentials", "true")
		w.Header().Add("Access-Control-Allow-Headers", "Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization, accept, origin, Cache-Control, X-Requested-With, X-Underbbs-Subscriber")
		w.Header().Add("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE")

		if r.Method == "OPTIONS" {
			http.Error(w, "No Content", http.StatusNoContent)
			return
		}

		next.ServeHTTP(w, r)
	})
}

func New(config models.GlobalSettings) *BBSServer {
	var apKey crypto.PrivateKey

	useApKey := false

	if config.ApKey != nil && config.ApDomain != nil {
		keybytes, err := ioutil.ReadFile(*config.ApKey)
		if err != nil {
			panic(err.Error())
		}
		keypem, _ := pem.Decode(keybytes)
		if keypem == nil {
			panic("couldn't decode pem block from AP key")
		}
		apKey, err = x509.ParsePKCS8PrivateKey(keypem.Bytes)
		if err != nil {
			panic(err.Error())
		}
		useApKey = true
	}

	var srvr *BBSServer

	if useApKey {
		srvr = &BBSServer{
			subscribeMessageBuffer: 16,
			logf:                   log.Printf,
			subscribers:            make(map[*Subscriber][]adapter.Adapter),
			apKey:                  &apKey,
			apDomain:               config.ApDomain,
		}
	} else {
		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("./frontend/dist")))

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

	// websocket
	srvr.serveMux.HandleFunc("/subscribe", srvr.subscribeHandler)
	// publish is unused currently, we just use the API and send data back on the websocket
	// 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{},
		OriginPatterns: []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() {
		self.logf("waiting for data on the subscriber's channel")
		for {
			select {
			case msg := <-s.msgs:
				writeTimeout(ctx, time.Second*5, c, msg)

			case <-ctx.Done():
				self.logf("subscriber has disconnected")
				for _, a := range self.subscribers[s] {
					// keeps any adapter's subscription from trying to send on the data channel after we close it
					a.Stop()
				}
				close(s.data)
				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)

	self.logf("data listener is done!")

	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)
}