add internal/{processor,streaming}

internal/streaming/event.go

@@ -0,0 +1,57 @@
+package streaming
+
+import "encoding/json"
+
+type EventType int8
+
+const (
+	EventTypeError EventType = 1
+	EventTypePost  EventType = 2
+
+	EventTypeSubscribe   EventType = 126
+	EventTypeUnsubscribe EventType = 127
+)
+
+func (et EventType) Valid() bool {
+	switch et {
+	case EventTypeError:
+		return true
+	case EventTypePost:
+		return true
+	case EventTypeSubscribe:
+		return true
+	case EventTypeUnsubscribe:
+		return true
+	default:
+		return false
+	}
+}
+
+// Returns true if this event can be subscribed to/unsubscribed from
+func (et EventType) ValidReceive() bool {
+	if !et.Valid() {
+		return false
+	}
+
+	switch et {
+	case EventTypeError, EventTypeSubscribe, EventTypeUnsubscribe:
+		return false
+	default:
+		return true
+	}
+}
+
+type Event struct {
+	Type EventType `json:"t"`
+	Data any       `json:"d"`
+}
+
+type ErrorEvent struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+type IncomingEvent struct {
+	Type EventType       `json:"t"`
+	Data json.RawMessage `json:"d"` // this is a RawMessage so we can easily unmarshal it later
+}

internal/streaming/sockets.go

@@ -0,0 +1,114 @@
+package streaming
+
+import (
+	"context"
+	"sync"
+
+	"github.com/oklog/ulid/v2"
+)
+
+type SocketHolder struct {
+	// map of sockets to
+	sockets map[ulid.ULID]*userSockets
+	mu      sync.Mutex
+}
+
+func (sh *SocketHolder) Send(acctID ulid.ULID, et EventType, data any) {
+	userSockets := sh.SocketsFor(acctID)
+
+	userSockets.mu.Lock()
+	sockets := make([]*Socket, len(userSockets.sockets))
+	copy(sockets, userSockets.sockets)
+	userSockets.mu.Unlock()
+
+	for _, s := range sockets {
+		if s.willAcceptEvent(et) {
+			// the socket might block for a bit, so spin this off into a separate goroutine
+			go func(s *Socket) {
+				s.ch <- Event{Type: et, Data: data}
+			}(s)
+		}
+	}
+}
+
+func (s *SocketHolder) SocketsFor(acct ulid.ULID) *userSockets {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	us, ok := s.sockets[acct]
+	if !ok {
+		us = &userSockets{}
+		s.sockets[acct] = us
+	}
+	return us
+}
+
+const sessionCountLimit = 50 // no more than 50 concurrent sessions per user
+
+type userSockets struct {
+	mu      sync.Mutex
+	sockets []*Socket
+}
+
+func (s *userSockets) NewSocket(ctx context.Context, cancel context.CancelFunc) (*Socket, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.sockets) >= sessionCountLimit {
+		return nil, false
+	}
+	socket := NewSocket(ctx, cancel)
+	s.sockets = append(s.sockets, socket)
+	return socket, true
+}
+
+type Socket struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	ch    chan Event
+	types map[EventType]struct{}
+
+	mu sync.RWMutex
+}
+
+func (s *Socket) willAcceptEvent(mt EventType) bool {
+	if mt == EventTypeError {
+		return true
+	}
+
+	s.mu.RLock()
+	_, ok := s.types[mt]
+	s.mu.RUnlock()
+	return ok
+}
+
+func (s *Socket) SetEvent(mt EventType, add bool) {
+	s.mu.Lock()
+	if add {
+		s.types[mt] = struct{}{}
+	} else {
+		delete(s.types, mt)
+	}
+	s.mu.Unlock()
+}
+
+func (s *Socket) Cancel() {
+	s.cancel()
+}
+
+func (s *Socket) Done() <-chan struct{} {
+	return s.ctx.Done()
+}
+
+func (s *Socket) Chan() <-chan Event {
+	return s.ch
+}
+
+func NewSocket(ctx context.Context, cancel context.CancelFunc) *Socket {
+	return &Socket{
+		ctx:    ctx,
+		cancel: cancel,
+		ch:     make(chan Event),
+		types:  make(map[EventType]struct{}),
+	}
+}