package gametable import ( "context" "errors" "golang.org/x/time/rate" "io/ioutil" "log" "net/http" "nhooyr.io/websocket" "nilfm.cc/git/felt/admin" "nilfm.cc/git/felt/models" "nilfm.cc/git/felt/mongodb" "nilfm.cc/git/quartzgun/auth" "nilfm.cc/git/quartzgun/cookie" "sync" "time" ) type Subscriber struct { msgs chan []byte closeSlow func() } type GameTableServer struct { subscribeMessageBuffer int publishLimiter *rate.Limiter logf func(f string, v ...interface{}) serveMux http.ServeMux subscribersLock sync.Mutex subscribers map[*Subscriber]models.TableKey dbAdapter mongodb.DbAdapter } func New(adapter mongodb.DbAdapter, udb auth.UserStore) *GameTableServer { srvr := &GameTableServer{ subscribeMessageBuffer: 16, logf: log.Printf, subscribers: make(map[*Subscriber]models.TableKey), publishLimiter: rate.NewLimiter(rate.Every(time.Millisecond*100), 8), dbAdapter: adapter, } srvr.serveMux.Handle("/table/", http.FileServer(http.Dir("./static"))) srvr.serveMux.Handle("/admin/", admin.CreateAdminInterface(udb, adapter)) 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, nil) if err != nil { self.logf("%v", err) return } 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") }, } tableName, tblNameErr := cookie.GetToken("tableName", r) tablePasscode, tblPassErr := cookie.GetToken("tablePasscode", r) if tblNameErr != nil { return tblNameErr } else if tblPassErr != nil { return tblPassErr } tableKey := models.TableKey{ Name: tableName, Passcode: tablePasscode, } if !self.dbAdapter.CheckTable(tableKey) { return errors.New("Table with matching key was not found on this server") } 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() // decode message and store in DB self.publishLimiter.Wait(context.Background()) for s := range self.subscribers { select { case s.msgs <- msg: default: go s.closeSlow() } } } func (self *GameTableServer) getCurrentState(tableKey models.TableKey) []byte { // get diceroll log, map, and token state // build into a []byte message return make([]byte, 1) } 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) }