// Package realtime is the hub's live connection with signed-in proxies: which IMVU account each client is using // (presence), and the community products it's wearing. Later, room channels. // // Protocol (JSON text messages over a WebSocket at /api/v1/realtime, authorized with the proxy's access token): // // proxy -> hub {"type": "session", "imvu_id": 123} the client signed in as this IMVU account // proxy -> hub {"type": "outfit", "items": [{"id", "name"}]} the community products it wears now // hub -> proxy {"type": "welcome"} | {"type": "error", "message": "..."} package realtime import ( "context" "encoding/json" "errors" "log" "net/http" "sort" "sync" "time" "github.com/coder/websocket" ) // Item is a community product someone wears. type Item struct { ID string `json:"id"` Name string `json:"name"` } // Online is an IMVU account in use in a T5DE client now. type Online struct { PersonID int64 IMVUID int64 Since time.Time Outfit []Item } // Owners tells the hub which person an IMVU account is linked to. type Owners interface { LinkOwner(ctx context.Context, imvuID int64) (personID int64, ok bool, err error) } // Hub holds the connected proxies. type Hub struct { owners Owners log *log.Logger mu sync.Mutex sessions map[*session]bool changed chan struct{} // closed and replaced on every change, for live views } type session struct { personID int64 conn *websocket.Conn imvuID int64 since time.Time outfit []Item } func New(owners Owners, logger *log.Logger) *Hub { return &Hub{owners: owners, log: logger, sessions: map[*session]bool{}, changed: make(chan struct{})} } // Changed is closed the next time anyone's presence or outfit changes. func (h *Hub) Changed() <-chan struct{} { h.mu.Lock() defer h.mu.Unlock() return h.changed } func (h *Hub) notifyLocked() { close(h.changed) h.changed = make(chan struct{}) } // Online lists the IMVU accounts in use now, one entry each (the most recent session), by IMVU ID. func (h *Hub) Online() []Online { h.mu.Lock() defer h.mu.Unlock() byID := map[int64]Online{} for s := range h.sessions { if s.imvuID == 0 { continue } if prev, ok := byID[s.imvuID]; ok && prev.Since.After(s.since) { continue } byID[s.imvuID] = Online{PersonID: s.personID, IMVUID: s.imvuID, Since: s.since, Outfit: append([]Item(nil), s.outfit...)} } out := make([]Online, 0, len(byID)) for _, o := range byID { out = append(out, o) } sort.Slice(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) }) return out } // Connections counts connected proxies (signed in or not to an IMVU account yet). func (h *Hub) Connections() int { h.mu.Lock() defer h.mu.Unlock() return len(h.sessions) } // Disconnect closes a person's connections (when they're banned or sign their proxies out). func (h *Hub) Disconnect(personID int64) { h.mu.Lock() var conns []*websocket.Conn for s := range h.sessions { if s.personID == personID { conns = append(conns, s.conn) } } h.mu.Unlock() for _, c := range conns { c.Close(websocket.StatusPolicyViolation, "signed out") } } type message struct { Type string `json:"type"` IMVUID int64 `json:"imvu_id,omitempty"` Items []Item `json:"items,omitempty"` Message string `json:"message,omitempty"` } // Serve runs one proxy's connection, already authorized as personID, until it closes. func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, personID int64) { conn, err := websocket.Accept(w, r, nil) if err != nil { return } conn.SetReadLimit(64 << 10) s := &session{personID: personID, conn: conn, since: time.Now()} h.mu.Lock() h.sessions[s] = true h.mu.Unlock() defer func() { h.mu.Lock() delete(h.sessions, s) if s.imvuID != 0 { h.notifyLocked() } h.mu.Unlock() conn.CloseNow() }() ctx := r.Context() send := func(m message) error { b, _ := json.Marshal(m) wctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() return conn.Write(wctx, websocket.MessageText, b) } if send(message{Type: "welcome"}) != nil { return } // Keep the connection alive through proxies and NATs. go func() { t := time.NewTicker(30 * time.Second) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: pctx, cancel := context.WithTimeout(ctx, 10*time.Second) err := conn.Ping(pctx) cancel() if err != nil { conn.CloseNow() return } } } }() for { _, data, err := conn.Read(ctx) if err != nil { return } var m message if json.Unmarshal(data, &m) != nil { send(message{Type: "error", Message: "not a message"}) continue } switch m.Type { case "session": owner, ok, err := h.owners.LinkOwner(ctx, m.IMVUID) if err != nil { send(message{Type: "error", Message: "try again later"}) continue } if !ok || owner != personID { // Not this person's account: no presence, no outfit for it. h.set(s, func() { s.imvuID, s.outfit = 0, nil }) send(message{Type: "error", Message: "not_linked"}) continue } h.set(s, func() { s.imvuID, s.since, s.outfit = m.IMVUID, time.Now(), nil }) case "outfit": if len(m.Items) > 64 { send(message{Type: "error", Message: "too many items"}) continue } items := make([]Item, 0, len(m.Items)) for _, it := range m.Items { if it.ID == "" || len(it.ID) > 128 || len(it.Name) > 128 { continue } items = append(items, it) } h.set(s, func() { s.outfit = items }) default: send(message{Type: "error", Message: "unknown message type"}) } } } func (h *Hub) set(s *session, change func()) { h.mu.Lock() defer h.mu.Unlock() change() h.notifyLocked() } // ErrUnauthorized is returned by Authorize funcs for a missing or bad token. var ErrUnauthorized = errors.New("unauthorized")