underbbs/server/server.go

package server

import (
  "forge.lightcrystal.systems/lightcrystal/underbbs/models"
	"hacklab.nilfm.cc/quartzgun/renderer"
	"net/http"
  "nhooyr.io/websocket"
  "encoding/base64"
  "encoding/json"
)


type Subscriber struct {
	msgs      chan []byte
	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][]models.Adapter
}

func New() *BBSServer {
  srvr := &BBSServer{
    subscribeMessageBuffer: 16,
    logf:                   log.Printf,
    subscribers:            make(map[*Subscriber][]model.Adapter),   
  }
  
  // frontend is here
	srvr.serveMux.Handle("/app/", http.StripPrefix("/app/", renderer.Subtree("./dist")))
  
	// websocket
	srvr.serveMux.HandleFunc("/subscribe", srvr.subscribeHandler)
	srvr.serveMux.HandleFunc("/publish", srvr.publishHandler)
	
	return srvr
}

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

func (self *GameTableServer) subscribeHandler(w http.ResponseWriter, r *http.Request) {
	
	c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
		Subprotocols: [],
	})
	if err != nil {
		self.logf("%v", err)
		return
	}
	
	// decode subprotocol data into settings objects
	data := c.Subprotocol()
	
	// base64 decode
	decoded, err := base64.StdEncoding.DecodeString(data)
	if err != nil {
	  c.Close(3000, err.Error())
	}
	
	settings := []models.Settings
	// unmarshal the json into the settings array
	err := json.Unmarshal([]byte(decoded), &settings)
	if (err != nil) {
	  c.Close(3001, err.Erorr())
	}

	// attempt to initialize adapters
	for i, s := range settings {
	  switch(s.Protocol) {
	    case "nostr":
	      break;
	    case "masto":
	      break;
	    default:
	      break;
	  }
	}
	// keep reference to the adapters?
	
	defer c.Close(websocket.StatusInternalError, "")

	err = self.subscribe(r, c)
	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 (self *GameTableServer) subscribe(r *http.Request, c *websocket.Conn) error {
	ctx := r.Context()
	ctx = c.CloseRead(ctx)

	s := &Subscriber{
		msgs: make(chan []byte, self.subscribeMessageBuffer),
		closeSlow: func() {
			c.Close(websocket.StatusPolicyViolation, "connection too slow to keep up with messages")
		},
	}
	
	// validate the subprotocol

	self.addSubscriber(s, tableKey)

	defer self.deleteSubscriber(s)
	select {
	case s.msgs <- self.getCurrentState(tableKey):
	default:
		go s.closeSlow()
	}

	for {
		select {
		case msg := <-s.msgs:
			err := writeTimeout(ctx, time.Second*5, c, msg)
			if err != nil {
				return err
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func (self *GameTableServer) 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 *GameTableServer) 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 *GameTableServer) addSubscriber(s *Subscriber, k models.TableKey) {
	self.subscribersLock.Lock()
	self.subscribers[s] = k
	self.subscribersLock.Unlock()
}

func (self *GameTableServer) 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)
}