package gametable import ( "bytes" "context" "encoding/json" "errors" "fmt" "golang.org/x/time/rate" "hacklab.nilfm.cc/felt/admin" "hacklab.nilfm.cc/felt/models" "hacklab.nilfm.cc/felt/mongodb" "hacklab.nilfm.cc/quartzgun/auth" "hacklab.nilfm.cc/quartzgun/renderer" "io/ioutil" "log" "net/http" "nhooyr.io/websocket" "strings" "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.StripPrefix("/table/", renderer.Subtree("./static"))) srvr.serveMux.Handle("/admin/", http.StripPrefix("/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) { protocols, err := self.dbAdapter.GetProtocols() if err != nil { self.logf("%v", err) return } c, err := websocket.Accept(w, r, &websocket.AcceptOptions{ Subprotocols: protocols, }) 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") }, } subprotoParts := strings.Split(c.Subprotocol(), ".") if len(subprotoParts) != 2 { return errors.New("Couldn't decode subprotocol") } tableKey := models.TableKey{ Name: subprotoParts[0], Passcode: subprotoParts[1], } // subprotocol is guaranteed to be an existing table, but maybe just leave this here 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 tableMsg := models.TableMessage{} err := json.NewDecoder(bytes.NewReader(msg)).Decode(&tableMsg) if err != nil { fmt.Println(err.Error()) return } self.writeToDB(tableMsg) self.publishLimiter.Wait(context.Background()) for s, k := range self.subscribers { if k == *tableMsg.Key { select { case s.msgs <- msg: default: go s.closeSlow() } } } } func (self *GameTableServer) getCurrentState(tableKey models.TableKey) []byte { // get diceroll log, map, and token state if self.dbAdapter.CheckTable(tableKey) { mapUrl, _ := self.dbAdapter.GetMapImageUrl(tableKey) auxMessage, _ := self.dbAdapter.GetAuxMessage(tableKey) availableTokens, _ := self.dbAdapter.GetTokens(tableKey, true) activeTokens, _ := self.dbAdapter.GetTokens(tableKey, false) diceRolls, _ := self.dbAdapter.GetDiceRolls(tableKey) table := models.Table{ Name: tableKey.Name, Passcode: tableKey.Passcode, DiceRolls: diceRolls, MapImageUrl: mapUrl, Tokens: activeTokens, AvailableTokens: availableTokens, AuxMessage: auxMessage, } data, err := json.Marshal(table) if err != nil { return []byte("{\"error\": \"" + err.Error() + "\"}") } else { return data } } return []byte("{\"error\": \"table not found\"}") } func (self *GameTableServer) writeToDB(tableMsg models.TableMessage) error { key := *tableMsg.Key if tableMsg.DiceRoll != nil { err := self.dbAdapter.InsertDiceRoll(key, *tableMsg.DiceRoll) if err != nil { fmt.Println(err.Error()) } } return nil } 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) }