Improves presence/activity handling over websocket

internal/payload/outgoing/activity.go

@@ -9,6 +9,7 @@ type (
 		Kind      string `json:"kind,omitempty"`
 		MessageID uint64 `json:"messageID,string,omitempty"`
 		ChannelID uint64 `json:"channelID,string,omitempty"`
+		Present   bool   `json:"present"`
 	}
 )
 

internal/payload/outgoing/connection.go

@@ -1,25 +0,0 @@
-package outgoing
-
-import (
-	"encoding/json"
-)
-
-type (
-	Connected struct {
-		// Who did connect?
-		UserID string `json:"uid"`
-	}
-
-	Disconnected struct {
-		// Who did disconnect?
-		UserID string `json:"uid"`
-	}
-)
-
-func (p *Connected) EncodeMessage() ([]byte, error) {
-	return json.Marshal(Payload{Connected: p})
-}
-
-func (p *Disconnected) EncodeMessage() ([]byte, error) {
-	return json.Marshal(Payload{Disconnected: p})
-}

internal/payload/outgoing/payload.go

@@ -4,9 +4,6 @@ type (
 	Payload struct {
 		*Error `json:"error,omitempty"`
 
-		*Connected    `json:"clientConnected,omitempty"`
-		*Disconnected `json:"clientDisconnected,omitempty"`
-
 		*Message    `json:"message,omitempty"`
 		*MessageSet `json:"messages,omitempty"`
 

messaging/rest/activity.go

@@ -29,6 +29,13 @@ func (ctrl *Activity) Send(ctx context.Context, r *request.ActivitySend) (interf
 		return nil, errors.New("can not send activity on message without channel ID")
 	}
 
+	switch r.Kind {
+	case "":
+		return nil, errors.New("missing value for activity kind")
+	case "connected", "disconnected":
+		return nil, errors.New("can not use reserved values for activity kind")
+	}
+
 	return true, ctrl.event.With(ctx).Activity(&types.Activity{
 		UserID:    auth.GetIdentityFromContext(ctx).Identity(),
 		ChannelID: r.ChannelID,

messaging/websocket/session.go

@@ -23,7 +23,9 @@ type (
 		id   uint64
 		once sync.Once
 		conn *websocket.Conn
-		ctx  context.Context
+
+		ctx       context.Context
+		ctxCancel context.CancelFunc
 
 		subs *Subscriptions
 
@@ -44,15 +46,17 @@ type (
 )
 
 func (Session) New(ctx context.Context, config *repository.Flags, conn *websocket.Conn) *Session {
+
 	s := &Session{
 		conn:   conn,
-		ctx:    ctx,
 		config: config,
 		subs:   NewSubscriptions(),
 		send:   make(chan []byte, 512),
 		stop:   make(chan []byte, 1),
 	}
 
+	s.ctx, s.ctxCancel = context.WithCancel(ctx)
+
 	s.svc.ch = service.DefaultChannel
 	s.svc.msg = service.DefaultMessage
 
@@ -86,16 +90,53 @@ func (sess *Session) connected() (err error) {
 	}
 
 	// Tell everyone that user has connected
-	if err = sess.sendToAll(&outgoing.Connected{UserID: payload.Uint64toa(sess.user.Identity())}); err != nil {
+	if err = sess.sendPresence("connected"); err != nil {
 		return
 	}
 
+	// Create a heartbeat every minute for this user
+	go func() {
+		t := time.NewTicker(time.Second * 3)
+		for {
+			select {
+			case <-sess.ctx.Done():
+				return
+			case <-t.C:
+				sess.sendPresence("")
+			}
+		}
+	}()
+
 	return nil
 }
 
 func (sess *Session) disconnected() {
 	// Tell everyone that user has disconnected
-	_ = sess.sendToAll(&outgoing.Disconnected{UserID: payload.Uint64toa(sess.user.Identity())})
+	_ = sess.sendPresence("disconnected")
+
+	// Cancel context
+	sess.ctxCancel()
+
+	// Close connection
+	sess.conn.Close()
+	sess.conn = nil
+}
+
+// Sends user presence information to all subscribers
+//
+// It sends "connected", "disconnected" and "" activity kinds
+func (sess *Session) sendPresence(kind string) error {
+	connections := store.CountConnections(sess.user.Identity())
+	if kind == "disconnected" {
+		connections--
+	}
+
+	// Tell everyone that user has disconnected
+	return sess.sendToAll(&outgoing.Activity{
+		UserID:  sess.user.Identity(),
+		Kind:    kind,
+		Present: connections > 0,
+	})
 }
 
 func (sess *Session) Handle() (err error) {
@@ -111,8 +152,6 @@ func (sess *Session) Handle() (err error) {
 func (sess *Session) Close() {
 	sess.once.Do(func() {
 		sess.disconnected()
-		sess.conn.Close()
-		sess.conn = nil
 		store.Delete(sess.id)
 	})
 }

messaging/websocket/session_store.go

@@ -1,8 +1,9 @@
 package websocket
 
 import (
-	"github.com/titpetric/factory"
 	"sync"
+
+	"github.com/titpetric/factory"
 )
 
 type (
@@ -39,6 +40,16 @@ func (s *Store) Walk(callback func(*Session)) {
 	}
 }
 
+func (s *Store) CountConnections(userID uint64) (count uint) {
+	s.Walk(func(session *Session) {
+		if session.user.Identity() == userID {
+			count++
+		}
+	})
+
+	return
+}
+
 func (s *Store) Get(id uint64) *Session {
 	s.RLock()
 	defer s.RUnlock()